diff --git a/apps/api/migrations/versions/0021_product_features.py b/apps/api/migrations/versions/0021_product_features.py
new file mode 100644
index 0000000..5a453e9
--- /dev/null
+++ b/apps/api/migrations/versions/0021_product_features.py
@@ -0,0 +1,154 @@
+"""Product descriptions and their confirmed technical features (ADR 0010).
+
+Revision ID: 0021_product_features
+Revises: 0020_claim_documents
+
+Customer-confidential data: tenant tables with FORCE RLS. Descriptions are
+append-only. A feature set is editable while "draft"; once "confirmed" the
+set and its features cannot be changed or deleted, enforced by triggers (the
+immutable approval boundary), not only by application code.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import JSONB
+
+from helpers.tenancy import enable_force_rls
+
+revision = "0021_product_features"
+down_revision = "0020_claim_documents"
+branch_labels = None
+depends_on = None
+
+RUNTIME_ROLES = ("patent_evidence_app", "patent_evidence_worker")
+TABLES = ("product_descriptions", "product_feature_sets", "product_features")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "product_descriptions",
+        sa.Column("id", sa.Uuid(), primary_key=True, nullable=False),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("case_id", sa.Uuid(), nullable=False),
+        sa.Column("version_number", sa.Integer(), nullable=False),
+        sa.Column("description_text", sa.Text(), nullable=False),
+        sa.Column("text_sha256", sa.String(64), nullable=False),
+        sa.Column("created_by_identity_id", sa.Uuid(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.ForeignKeyConstraint(["organization_id"], ["organizations.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(["case_id"], ["cases.id"], ondelete="CASCADE"),
+        sa.UniqueConstraint("case_id", "version_number", name="uq_product_descriptions_case_version"),
+        sa.CheckConstraint("text_sha256 ~ '^[0-9a-f]{64}$'", name="ck_product_descriptions_sha256"),
+        sa.CheckConstraint("length(description_text) > 0", name="ck_product_descriptions_not_empty"),
+    )
+
+    op.create_table(
+        "product_feature_sets",
+        sa.Column("id", sa.Uuid(), primary_key=True, nullable=False),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("case_id", sa.Uuid(), nullable=False),
+        sa.Column("description_id", sa.Uuid(), nullable=False),
+        sa.Column("version_number", sa.Integer(), nullable=False),
+        sa.Column("parent_set_id", sa.Uuid(), nullable=True),
+        sa.Column("status", sa.String(16), nullable=False, server_default="draft"),
+        sa.Column("splitter_version", sa.String(64), nullable=False),
+        sa.Column("created_by_identity_id", sa.Uuid(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("confirmed_by_identity_id", sa.Uuid(), nullable=True),
+        sa.Column("confirmed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["organization_id"], ["organizations.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(["case_id"], ["cases.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["description_id"], ["product_descriptions.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(["parent_set_id"], ["product_feature_sets.id"], ondelete="RESTRICT"),
+        sa.UniqueConstraint("case_id", "version_number", name="uq_product_feature_sets_case_version"),
+        sa.CheckConstraint("status IN ('draft', 'confirmed')", name="ck_product_feature_sets_status"),
+        sa.CheckConstraint(
+            "(status = 'confirmed') = (confirmed_by_identity_id IS NOT NULL AND confirmed_at IS NOT NULL)",
+            name="ck_product_feature_sets_confirmation",
+        ),
+    )
+
+    op.create_table(
+        "product_features",
+        sa.Column("id", sa.Uuid(), primary_key=True, nullable=False),
+        sa.Column("organization_id", sa.Uuid(), nullable=False),
+        sa.Column("case_id", sa.Uuid(), nullable=False),
+        sa.Column("feature_set_id", sa.Uuid(), nullable=False),
+        sa.Column("feature_code", sa.String(16), nullable=False),
+        sa.Column("feature_text", sa.Text(), nullable=False),
+        # [[start, end], ...] offsets into the description text; [] for manual features.
+        sa.Column("spans", JSONB(), nullable=False, server_default=sa.text("'[]'::jsonb")),
+        sa.Column("origin", sa.String(8), nullable=False),
+        sa.Column("sort_order", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.ForeignKeyConstraint(["organization_id"], ["organizations.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(["case_id"], ["cases.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["feature_set_id"], ["product_feature_sets.id"], ondelete="CASCADE"),
+        sa.UniqueConstraint("feature_set_id", "feature_code", name="uq_product_features_set_code"),
+        sa.CheckConstraint("origin IN ('split', 'manual', 'edited')", name="ck_product_features_origin"),
+        sa.CheckConstraint("length(feature_text) > 0", name="ck_product_features_not_empty"),
+    )
+    op.create_index("ix_product_features_set", "product_features", ["feature_set_id", "sort_order"])
+
+    op.execute(
+        """CREATE FUNCTION deny_confirmed_product_feature_set_change() RETURNS trigger AS $$
+        BEGIN
+          IF OLD.status = 'confirmed' THEN
+            RAISE EXCEPTION 'confirmed product feature set % is immutable', OLD.id;
+          END IF;
+          IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
+          RETURN NEW;
+        END; $$ LANGUAGE plpgsql"""
+    )
+    op.execute(
+        """CREATE TRIGGER product_feature_sets_confirmed_immutable
+        BEFORE UPDATE OR DELETE ON product_feature_sets
+        FOR EACH ROW EXECUTE FUNCTION deny_confirmed_product_feature_set_change()"""
+    )
+    op.execute(
+        """CREATE FUNCTION deny_features_of_confirmed_set() RETURNS trigger AS $$
+        DECLARE
+          target uuid;
+        BEGIN
+          FOR target IN
+            SELECT unnest(ARRAY[
+              CASE WHEN TG_OP IN ('UPDATE', 'DELETE') THEN OLD.feature_set_id END,
+              CASE WHEN TG_OP IN ('INSERT', 'UPDATE') THEN NEW.feature_set_id END])
+          LOOP
+            IF target IS NOT NULL AND EXISTS (
+              SELECT 1 FROM product_feature_sets WHERE id = target AND status = 'confirmed'
+            ) THEN
+              RAISE EXCEPTION 'features of confirmed product feature set % are immutable', target;
+            END IF;
+          END LOOP;
+          IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
+          RETURN NEW;
+        END; $$ LANGUAGE plpgsql"""
+    )
+    op.execute(
+        """CREATE TRIGGER product_features_confirmed_immutable
+        BEFORE INSERT OR UPDATE OR DELETE ON product_features
+        FOR EACH ROW EXECUTE FUNCTION deny_features_of_confirmed_set()"""
+    )
+    for table in TABLES:
+        op.execute(
+            f"CREATE TRIGGER {table}_organization_immutable BEFORE UPDATE ON {table} "
+            "FOR EACH ROW EXECUTE FUNCTION deny_organization_id_change()"
+        )
+
+    for role in RUNTIME_ROLES:
+        op.execute(f'GRANT SELECT, INSERT ON "product_descriptions" TO {role}')
+        op.execute(f'GRANT SELECT, INSERT, UPDATE, DELETE ON "product_feature_sets" TO {role}')
+        op.execute(f'GRANT SELECT, INSERT, UPDATE, DELETE ON "product_features" TO {role}')
+    for table in TABLES:
+        enable_force_rls(table)
+
+
+def downgrade() -> None:
+    op.drop_table("product_features")
+    op.drop_table("product_feature_sets")
+    op.drop_table("product_descriptions")
+    op.execute("DROP FUNCTION deny_features_of_confirmed_set()")
+    op.execute("DROP FUNCTION deny_confirmed_product_feature_set_change()")
diff --git a/apps/api/src/patent_evidence_api/fto/__init__.py b/apps/api/src/patent_evidence_api/fto/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/apps/api/src/patent_evidence_api/fto/product_features.py b/apps/api/src/patent_evidence_api/fto/product_features.py
new file mode 100644
index 0000000..91406c8
--- /dev/null
+++ b/apps/api/src/patent_evidence_api/fto/product_features.py
@@ -0,0 +1,335 @@
+"""Product descriptions and their technical features (ADR 0010).
+
+Descriptions are append-only versions. A feature set starts as a draft from
+the deterministic splitter, is edited by a person (edit, add, delete, split,
+merge) and is then confirmed; confirmed sets are immutable (database triggers)
+and change only through a new revision. All tables are tenant-scoped (RLS).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from collections.abc import Callable
+from datetime import datetime
+from typing import Any
+from uuid import UUID, uuid4
+
+from fastapi import HTTPException
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from modules.fto.product import SPLITTER_VERSION, split_description
+
+MAX_DESCRIPTION_CHARS = 50_000
+
+
+class ProductFeatureService:
+    def __init__(self, clock: Callable[[], datetime]) -> None:
+        self._clock = clock
+
+    # ------------------------------------------------------------ descriptions
+
+    async def add_description(
+        self, session: AsyncSession, *, organization_id: UUID, case_id: UUID, description: str, actor_id: UUID
+    ) -> dict[str, Any]:
+        body = description.strip()
+        if not body:
+            raise HTTPException(status_code=422, detail="product_description_empty")
+        if len(body) > MAX_DESCRIPTION_CHARS:
+            raise HTTPException(status_code=422, detail="product_description_too_long")
+        await self._require_case(session, organization_id, case_id)
+        version = await self._next_version(session, "product_descriptions", case_id)
+        description_id = uuid4()
+        await session.execute(
+            text(
+                """INSERT INTO product_descriptions
+                (id, organization_id, case_id, version_number, description_text, text_sha256,
+                 created_by_identity_id, created_at)
+                VALUES (:id, :org, :case, :version, :text, :sha, :actor, :now)"""
+            ),
+            {
+                "id": description_id, "org": organization_id, "case": case_id, "version": version,
+                "text": body, "sha": hashlib.sha256(body.encode("utf-8")).hexdigest(),
+                "actor": actor_id, "now": self._clock(),
+            },
+        )
+        return {"id": str(description_id), "version_number": version}
+
+    # ------------------------------------------------------------ feature sets
+
+    async def create_draft(
+        self, session: AsyncSession, *, organization_id: UUID, case_id: UUID, description_id: UUID, actor_id: UUID
+    ) -> dict[str, Any]:
+        row = (
+            await session.execute(
+                text(
+                    """SELECT description_text FROM product_descriptions
+                    WHERE id = :id AND case_id = :case AND organization_id = :org"""
+                ),
+                {"id": description_id, "case": case_id, "org": organization_id},
+            )
+        ).one_or_none()
+        if row is None:
+            raise HTTPException(status_code=404, detail="product_description_not_found")
+        result = split_description(row[0])
+        set_id = await self._insert_set(session, organization_id, case_id, description_id, actor_id, parent=None)
+        for order, feature in enumerate(result.features):
+            await self._insert_feature(
+                session, organization_id, case_id, set_id, feature.feature_id, feature.text,
+                [list(s) for s in feature.spans], "split", order,
+            )
+        return {**(await self.get_set(session, organization_id=organization_id, set_id=set_id)), "warnings": list(result.warnings)}
+
+    async def create_revision(
+        self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, actor_id: UUID
+    ) -> dict[str, Any]:
+        source = await self._load_set(session, organization_id, set_id)
+        if source["status"] != "confirmed":
+            raise HTTPException(status_code=409, detail="only_confirmed_sets_can_be_revised")
+        new_id = await self._insert_set(
+            session, organization_id, source["case_id"], source["description_id"], actor_id, parent=set_id
+        )
+        for feature in source["features"]:
+            await self._insert_feature(
+                session, organization_id, source["case_id"], new_id, feature["code"], feature["text"],
+                feature["spans"], feature["origin"], feature["sort_order"],
+            )
+        return await self.get_set(session, organization_id=organization_id, set_id=new_id)
+
+    async def confirm(self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, actor_id: UUID) -> dict[str, Any]:
+        current = await self._load_draft(session, organization_id, set_id)
+        if not current["features"]:
+            raise HTTPException(status_code=422, detail="cannot_confirm_empty_feature_set")
+        now = self._clock()
+        await session.execute(
+            text(
+                """UPDATE product_feature_sets
+                SET status = 'confirmed', confirmed_by_identity_id = :actor, confirmed_at = :now, updated_at = :now
+                WHERE id = :id AND organization_id = :org"""
+            ),
+            {"actor": actor_id, "now": now, "id": set_id, "org": organization_id},
+        )
+        return await self.get_set(session, organization_id=organization_id, set_id=set_id)
+
+    # ------------------------------------------------------------ draft edits
+
+    async def edit_feature(
+        self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, code: str, new_text: str
+    ) -> dict[str, Any]:
+        draft = await self._load_draft(session, organization_id, set_id)
+        feature = self._feature(draft, code)
+        body = new_text.strip()
+        if not body:
+            raise HTTPException(status_code=422, detail="feature_text_empty")
+        origin = feature["origin"] if body == feature["text"] else ("manual" if feature["origin"] == "manual" else "edited")
+        await self._update_feature(session, organization_id, set_id, code, body, feature["spans"], origin)
+        return await self.get_set(session, organization_id=organization_id, set_id=set_id)
+
+    async def add_feature(
+        self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, feature_text: str
+    ) -> dict[str, Any]:
+        draft = await self._load_draft(session, organization_id, set_id)
+        body = feature_text.strip()
+        if not body:
+            raise HTTPException(status_code=422, detail="feature_text_empty")
+        order = max((f["sort_order"] for f in draft["features"]), default=-1) + 1
+        await self._insert_feature(
+            session, organization_id, draft["case_id"], set_id, self._next_code(draft), body, [], "manual", order
+        )
+        return await self.get_set(session, organization_id=organization_id, set_id=set_id)
+
+    async def delete_feature(self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, code: str) -> dict[str, Any]:
+        draft = await self._load_draft(session, organization_id, set_id)
+        self._feature(draft, code)
+        await session.execute(
+            text("DELETE FROM product_features WHERE feature_set_id = :set AND feature_code = :code AND organization_id = :org"),
+            {"set": set_id, "code": code, "org": organization_id},
+        )
+        return await self.get_set(session, organization_id=organization_id, set_id=set_id)
+
+    async def split_feature(
+        self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, code: str, at: int
+    ) -> dict[str, Any]:
+        """Split one feature's text at character index ``at`` into two features."""
+        draft = await self._load_draft(session, organization_id, set_id)
+        feature = self._feature(draft, code)
+        left, right = feature["text"][:at].strip(), feature["text"][at:].strip()
+        if not left or not right:
+            raise HTTPException(status_code=422, detail="split_would_create_empty_feature")
+        left_spans, right_spans = feature["spans"], []
+        # Unedited, contiguous split features keep exact spans on both sides.
+        if feature["origin"] == "split" and len(feature["spans"]) == 1:
+            start, end = feature["spans"][0]
+            raw_left, raw_right = feature["text"][:at], feature["text"][at:]
+            left_end = start + len(raw_left.rstrip())
+            right_start = start + at + (len(raw_right) - len(raw_right.lstrip()))
+            left_spans, right_spans = [[start, left_end]], [[right_start, end]]
+            left_origin = right_origin = "split"
+        else:
+            left_origin = right_origin = "edited" if feature["origin"] != "manual" else "manual"
+        await self._update_feature(session, organization_id, set_id, code, left, left_spans, left_origin)
+        await self._shift_orders_after(session, organization_id, set_id, feature["sort_order"])
+        await self._insert_feature(
+            session, organization_id, draft["case_id"], set_id, self._next_code(draft), right,
+            right_spans, right_origin, feature["sort_order"] + 1,
+        )
+        return await self.get_set(session, organization_id=organization_id, set_id=set_id)
+
+    async def merge_features(
+        self, session: AsyncSession, *, organization_id: UUID, set_id: UUID, codes: list[str]
+    ) -> dict[str, Any]:
+        """Merge features (in their current order) into the first of them."""
+        draft = await self._load_draft(session, organization_id, set_id)
+        if len(set(codes)) < 2:
+            raise HTTPException(status_code=422, detail="merge_needs_two_features")
+        chosen = sorted((self._feature(draft, c) for c in set(codes)), key=lambda f: f["sort_order"])
+        first, rest = chosen[0], chosen[1:]
+        merged_text = " ".join(f["text"] for f in chosen)
+        merged_spans = [s for f in chosen for s in f["spans"]]
+        origin = "split" if all(f["origin"] == "split" for f in chosen) else "edited"
+        await self._update_feature(session, organization_id, set_id, first["code"], merged_text, merged_spans, origin)
+        for feature in rest:
+            await session.execute(
+                text("DELETE FROM product_features WHERE feature_set_id = :set AND feature_code = :code AND organization_id = :org"),
+                {"set": set_id, "code": feature["code"], "org": organization_id},
+            )
+        return await self.get_set(session, organization_id=organization_id, set_id=set_id)
+
+    # ------------------------------------------------------------ reads
+
+    async def get_set(self, session: AsyncSession, *, organization_id: UUID, set_id: UUID) -> dict[str, Any]:
+        data = await self._load_set(session, organization_id, set_id)
+        return {
+            "id": str(data["id"]), "case_id": str(data["case_id"]), "description_id": str(data["description_id"]),
+            "version_number": data["version_number"], "status": data["status"],
+            "parent_set_id": str(data["parent_set_id"]) if data["parent_set_id"] else None,
+            "splitter_version": data["splitter_version"],
+            "confirmed_at": data["confirmed_at"].isoformat() if data["confirmed_at"] else None,
+            "features": [
+                {"code": f["code"], "text": f["text"], "spans": f["spans"], "origin": f["origin"]}
+                for f in data["features"]
+            ],
+        }
+
+    # ------------------------------------------------------------ internals
+
+    async def _require_case(self, session: AsyncSession, organization_id: UUID, case_id: UUID) -> None:
+        found = (
+            await session.execute(
+                text("SELECT 1 FROM cases WHERE id = :case AND organization_id = :org"),
+                {"case": case_id, "org": organization_id},
+            )
+        ).one_or_none()
+        if found is None:
+            raise HTTPException(status_code=404, detail="case_not_found")
+
+    async def _next_version(self, session: AsyncSession, table: str, case_id: UUID) -> int:
+        current = (
+            await session.execute(text(f"SELECT max(version_number) FROM {table} WHERE case_id = :case"), {"case": case_id})
+        ).scalar_one_or_none()
+        return (current or 0) + 1
+
+    async def _insert_set(
+        self, session: AsyncSession, organization_id: UUID, case_id: UUID, description_id: UUID, actor_id: UUID, *, parent: UUID | None
+    ) -> UUID:
+        set_id = uuid4()
+        now = self._clock()
+        await session.execute(
+            text(
+                """INSERT INTO product_feature_sets
+                (id, organization_id, case_id, description_id, version_number, parent_set_id, status,
+                 splitter_version, created_by_identity_id, created_at, updated_at)
+                VALUES (:id, :org, :case, :desc, :version, :parent, 'draft', :splitter, :actor, :now, :now)"""
+            ),
+            {
+                "id": set_id, "org": organization_id, "case": case_id, "desc": description_id,
+                "version": await self._next_version(session, "product_feature_sets", case_id),
+                "parent": parent, "splitter": SPLITTER_VERSION, "actor": actor_id, "now": now,
+            },
+        )
+        return set_id
+
+    async def _insert_feature(
+        self, session: AsyncSession, organization_id: UUID, case_id: UUID, set_id: UUID,
+        code: str, body: str, spans: list[list[int]], origin: str, order: int,
+    ) -> None:
+        now = self._clock()
+        await session.execute(
+            text(
+                """INSERT INTO product_features
+                (id, organization_id, case_id, feature_set_id, feature_code, feature_text, spans, origin,
+                 sort_order, created_at, updated_at)
+                VALUES (:id, :org, :case, :set, :code, :text, CAST(:spans AS jsonb), :origin, :order, :now, :now)"""
+            ),
+            {
+                "id": uuid4(), "org": organization_id, "case": case_id, "set": set_id, "code": code,
+                "text": body, "spans": json.dumps(spans), "origin": origin, "order": order, "now": now,
+            },
+        )
+
+    async def _update_feature(
+        self, session: AsyncSession, organization_id: UUID, set_id: UUID, code: str, body: str, spans: list, origin: str
+    ) -> None:
+        await session.execute(
+            text(
+                """UPDATE product_features
+                SET feature_text = :text, spans = CAST(:spans AS jsonb), origin = :origin, updated_at = :now
+                WHERE feature_set_id = :set AND feature_code = :code AND organization_id = :org"""
+            ),
+            {"text": body, "spans": json.dumps(spans), "origin": origin, "now": self._clock(),
+             "set": set_id, "code": code, "org": organization_id},
+        )
+
+    async def _shift_orders_after(self, session: AsyncSession, organization_id: UUID, set_id: UUID, order: int) -> None:
+        await session.execute(
+            text(
+                """UPDATE product_features SET sort_order = sort_order + 1
+                WHERE feature_set_id = :set AND organization_id = :org AND sort_order > :order"""
+            ),
+            {"set": set_id, "org": organization_id, "order": order},
+        )
+
+    async def _load_set(self, session: AsyncSession, organization_id: UUID, set_id: UUID) -> dict[str, Any]:
+        row = (
+            await session.execute(
+                text(
+                    """SELECT id, case_id, description_id, version_number, status, parent_set_id,
+                              splitter_version, confirmed_at
+                    FROM product_feature_sets WHERE id = :id AND organization_id = :org"""
+                ),
+                {"id": set_id, "org": organization_id},
+            )
+        ).mappings().one_or_none()
+        if row is None:
+            raise HTTPException(status_code=404, detail="product_feature_set_not_found")
+        features = (
+            await session.execute(
+                text(
+                    """SELECT feature_code AS code, feature_text AS text, spans, origin, sort_order
+                    FROM product_features WHERE feature_set_id = :id AND organization_id = :org
+                    ORDER BY sort_order, feature_code"""
+                ),
+                {"id": set_id, "org": organization_id},
+            )
+        ).mappings().all()
+        return {**dict(row), "features": [dict(f) for f in features]}
+
+    async def _load_draft(self, session: AsyncSession, organization_id: UUID, set_id: UUID) -> dict[str, Any]:
+        data = await self._load_set(session, organization_id, set_id)
+        if data["status"] != "draft":
+            raise HTTPException(status_code=409, detail="product_feature_set_confirmed")
+        return data
+
+    @staticmethod
+    def _feature(data: dict[str, Any], code: str) -> dict[str, Any]:
+        for feature in data["features"]:
+            if feature["code"] == code:
+                return feature
+        raise HTTPException(status_code=404, detail="product_feature_not_found")
+
+    @staticmethod
+    def _next_code(data: dict[str, Any]) -> str:
+        numbers = [int(m.group(1)) for f in data["features"] if (m := re.fullmatch(r"P(\d+)", f["code"]))]
+        return f"P{max(numbers, default=0) + 1}"
diff --git a/apps/api/tests/integration/test_product_features.py b/apps/api/tests/integration/test_product_features.py
new file mode 100644
index 0000000..0f4eb44
--- /dev/null
+++ b/apps/api/tests/integration/test_product_features.py
@@ -0,0 +1,178 @@
+"""Product descriptions and features (ADR 0010): RLS, immutability triggers, edit/confirm flow."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterator
+from datetime import UTC, datetime
+from uuid import UUID, uuid4
+
+import psycopg
+import pytest
+from fastapi import HTTPException
+from test_support import (
+    BASELINE_IDENTITY_A,
+    BASELINE_ORG_A,
+    BASELINE_ORG_B,
+    async_postgres_url,
+    postgres_url,
+    reset_database,
+    seed_two_tenant_baseline,
+)
+
+from patent_evidence_api.core.database import create_application_session_factory, create_engine, tenant_transaction
+from patent_evidence_api.fto.product_features import ProductFeatureService
+
+ORG_A, ORG_B, ACTOR = UUID(BASELINE_ORG_A), UUID(BASELINE_ORG_B), UUID(BASELINE_IDENTITY_A)
+CASE_A = UUID("70000000-0000-4000-8000-0000000000aa")
+NOW = datetime(2026, 10, 4, 8, 0, tzinfo=UTC)
+SYNTHETIC = "1. 液冷板由铝合金挤压成型，厚度为2.5 mm。\n2. 冷却液流道呈蛇形布置；流道入口与出口位于同一侧。\n3. 液冷板与电芯之间设置导热垫。"
+
+
+@pytest.fixture
+def migration() -> Iterator[psycopg.Connection]:
+    with psycopg.connect(
+        postgres_url("PE_TEST_MIGRATION_DATABASE_URL", "patent_evidence_migration"), autocommit=True
+    ) as connection:
+        reset_database(connection)
+        seed_two_tenant_baseline(connection, NOW)
+        connection.execute(
+            """INSERT INTO cases (id, organization_id, case_number, title, technical_field, created_by_identity_id)
+            VALUES (%s, %s, 'FTO-1', '液冷板 FTO（合成）', '电池热管理', %s)""",
+            (CASE_A, ORG_A, ACTOR),
+        )
+        yield connection
+
+
+def _run(organization: UUID, action):
+    async def go():
+        engine = create_engine(async_postgres_url("PE_TEST_APPLICATION_DATABASE_URL", "patent_evidence_app"))
+        try:
+            async with tenant_transaction(create_application_session_factory(engine), organization, actor_identity_id=ACTOR) as session:
+                return await action(session, ProductFeatureService(lambda: NOW))
+        finally:
+            await engine.dispose()
+
+    return asyncio.run(go())
+
+
+def _draft() -> dict:
+    async def action(session, service):
+        description = await service.add_description(
+            session, organization_id=ORG_A, case_id=CASE_A, description=SYNTHETIC, actor_id=ACTOR
+        )
+        return await service.create_draft(
+            session, organization_id=ORG_A, case_id=CASE_A, description_id=UUID(description["id"]), actor_id=ACTOR
+        )
+
+    return _run(ORG_A, action)
+
+
+def test_draft_from_description_has_exact_spans(migration: psycopg.Connection) -> None:
+    draft = _draft()
+    assert draft["status"] == "draft" and draft["version_number"] == 1
+    assert [f["text"] for f in draft["features"]] == [
+        "液冷板由铝合金挤压成型，厚度为2.5 mm。",
+        "冷却液流道呈蛇形布置；",
+        "流道入口与出口位于同一侧。",
+        "液冷板与电芯之间设置导热垫。",
+    ]
+    for feature in draft["features"]:
+        (start, end), = feature["spans"]
+        assert SYNTHETIC[start:end] == feature["text"]
+
+
+def test_other_tenant_cannot_see_product_data(migration: psycopg.Connection) -> None:
+    draft = _draft()
+
+    async def peek(session, service):
+        return await service.get_set(session, organization_id=ORG_B, set_id=UUID(draft["id"]))
+
+    with pytest.raises(HTTPException) as missing:
+        _run(ORG_B, peek)
+    assert missing.value.status_code == 404
+    with psycopg.connect(postgres_url("PE_TEST_APPLICATION_DATABASE_URL", "patent_evidence_app"), autocommit=True) as app:
+        app.execute("SELECT set_config('app.current_organization_id', %s, false)", (str(ORG_B),))
+        assert app.execute("SELECT count(*) FROM product_descriptions").fetchone() == (0,)
+        assert app.execute("SELECT count(*) FROM product_features").fetchone() == (0,)
+
+
+def test_edit_split_merge_add_delete_then_confirm(migration: psycopg.Connection) -> None:
+    draft = _draft()
+    set_id = UUID(draft["id"])
+
+    async def edits(session, service):
+        await service.merge_features(session, organization_id=ORG_A, set_id=set_id, codes=["P2", "P3"])
+        await service.split_feature(session, organization_id=ORG_A, set_id=set_id, code="P1", at=12)  # just after the comma (index 11)
+        await service.edit_feature(session, organization_id=ORG_A, set_id=set_id, code="P4", new_text="液冷板与电芯底面之间设置导热垫。")
+        await service.add_feature(session, organization_id=ORG_A, set_id=set_id, feature_text="箱体为钢制。")
+        return await service.delete_feature(session, organization_id=ORG_A, set_id=set_id, code="P6")
+
+    result = _run(ORG_A, edits)
+    by_code = {f["code"]: f for f in result["features"]}
+    assert by_code["P2"]["text"] == "冷却液流道呈蛇形布置； 流道入口与出口位于同一侧。" and by_code["P2"]["origin"] == "split"
+    assert by_code["P1"]["text"] == "液冷板由铝合金挤压成型，" and by_code["P1"]["origin"] == "split"
+    (s, e), = by_code["P5"]["spans"]  # right half of the split keeps an exact span
+    assert SYNTHETIC[s:e] == by_code["P5"]["text"] == "厚度为2.5 mm。"
+    assert by_code["P4"]["origin"] == "edited"
+    assert "P6" not in by_code  # the manual feature added above was deleted
+
+    async def confirm(session, service):
+        return await service.confirm(session, organization_id=ORG_A, set_id=set_id, actor_id=ACTOR)
+
+    assert _run(ORG_A, confirm)["status"] == "confirmed"
+
+    async def late_edit(session, service):
+        return await service.add_feature(session, organization_id=ORG_A, set_id=set_id, feature_text="不应成功")
+
+    with pytest.raises(HTTPException) as conflict:
+        _run(ORG_A, late_edit)
+    assert conflict.value.status_code == 409
+
+    async def revise(session, service):
+        return await service.create_revision(session, organization_id=ORG_A, set_id=set_id, actor_id=ACTOR)
+
+    revision = _run(ORG_A, revise)
+    assert revision["status"] == "draft" and revision["version_number"] == 2
+    assert revision["parent_set_id"] == str(set_id)
+    assert [f["code"] for f in revision["features"]] == [f["code"] for f in result["features"]]
+
+
+def test_database_enforces_the_approval_boundary(migration: psycopg.Connection) -> None:
+    """Even bypassing the service, a confirmed set and its features cannot change."""
+    draft = _draft()
+    set_id = UUID(draft["id"])
+    _run(ORG_A, lambda session, service: service.confirm(session, organization_id=ORG_A, set_id=set_id, actor_id=ACTOR))
+    with psycopg.connect(postgres_url("PE_TEST_APPLICATION_DATABASE_URL", "patent_evidence_app"), autocommit=True) as app:
+        app.execute("SELECT set_config('app.current_organization_id', %s, false)", (str(ORG_A),))
+        for statement in (
+            "UPDATE product_feature_sets SET status = 'draft' WHERE id = %s",
+            "DELETE FROM product_feature_sets WHERE id = %s",
+            "UPDATE product_features SET feature_text = 'changed' WHERE feature_set_id = %s",
+            "DELETE FROM product_features WHERE feature_set_id = %s",
+        ):
+            with pytest.raises(psycopg.errors.RaiseException):
+                app.execute(statement, (set_id,))
+        with pytest.raises(psycopg.errors.RaiseException):
+            app.execute(
+                """INSERT INTO product_features (id, organization_id, case_id, feature_set_id, feature_code,
+                   feature_text, origin, sort_order) VALUES (%s, %s, %s, %s, 'P99', 'x', 'manual', 99)""",
+                (uuid4(), ORG_A, CASE_A, set_id),
+            )
+        # Descriptions are append-only for runtime roles.
+        with pytest.raises(psycopg.errors.InsufficientPrivilege):
+            app.execute("UPDATE product_descriptions SET description_text = 'x'")
+
+
+def test_empty_set_cannot_be_confirmed(migration: psycopg.Connection) -> None:
+    draft = _draft()
+    set_id = UUID(draft["id"])
+
+    async def empty_then_confirm(session, service):
+        for feature in draft["features"]:
+            await service.delete_feature(session, organization_id=ORG_A, set_id=set_id, code=feature["code"])
+        return await service.confirm(session, organization_id=ORG_A, set_id=set_id, actor_id=ACTOR)
+
+    with pytest.raises(HTTPException) as invalid:
+        _run(ORG_A, empty_then_confirm)
+    assert invalid.value.status_code == 422
diff --git a/apps/api/tests/unit/test_fto_product_splitter.py b/apps/api/tests/unit/test_fto_product_splitter.py
new file mode 100644
index 0000000..63098cc
--- /dev/null
+++ b/apps/api/tests/unit/test_fto_product_splitter.py
@@ -0,0 +1,68 @@
+"""Product description splitter (ADR 0010). All texts are SYNTHETIC, written for these tests."""
+
+from __future__ import annotations
+
+from modules.fto.product import LONG_FEATURE_CHARS, SPLITTER_VERSION, split_description
+
+ZH = (
+    "电池包液冷板（合成示例）：\n"
+    "1. 液冷板由铝合金6063挤压成型，厚度为2.5 mm。\n"
+    "2、冷却液流道呈蛇形布置；流道入口与出口位于同一侧。\n"
+    "（3）液冷板与电芯底面之间设置导热垫，导热系数为1.5 W/(m·K)。\n"
+    "一、 电池模组通过螺栓固定在箱体上。"
+)
+
+
+def _texts(text: str) -> list[str]:
+    return [f.text for f in split_description(text).features]
+
+
+def test_chinese_numbered_list_and_clause_ends() -> None:
+    assert _texts(ZH) == [
+        "电池包液冷板（合成示例）：",
+        "液冷板由铝合金6063挤压成型，厚度为2.5 mm。",
+        "冷却液流道呈蛇形布置；",
+        "流道入口与出口位于同一侧。",
+        "液冷板与电芯底面之间设置导热垫，导热系数为1.5 W/(m·K)。",
+        "电池模组通过螺栓固定在箱体上。",
+    ]
+
+
+def test_spans_point_exactly_into_the_original() -> None:
+    result = split_description(ZH)
+    for feature in result.features:
+        (start, end), = feature.spans
+        assert ZH[start:end] == feature.text
+    assert [f.feature_id for f in result.features] == ["P1", "P2", "P3", "P4", "P5", "P6"]
+    assert result.splitter_version == SPLITTER_VERSION
+
+
+def test_english_sentences_keep_decimals_and_abbreviations() -> None:
+    text = "The plate is aluminium, e.g. 6063. The channel depth is 0.8 mm. See Fig. 3 for details; the pad is 1.0 mm."
+    assert _texts(text) == [
+        "The plate is aluminium, e.g. 6063.",
+        "The channel depth is 0.8 mm.",
+        "See Fig. 3 for details;",
+        "the pad is 1.0 mm.",
+    ]
+
+
+def test_bullets_stripped_but_signs_and_decimals_kept() -> None:
+    assert _texts("- Busbar made of copper\n• Insulating film\n-5 °C minimum temperature\n2.5 mm plate") == [
+        "Busbar made of copper",
+        "Insulating film",
+        "-5 °C minimum temperature",
+        "2.5 mm plate",
+    ]
+
+
+def test_chinese_comma_does_not_split() -> None:
+    assert _texts("导热垫设置在液冷板与电芯之间，厚度为1 mm，硬度为邵氏A 30。") == [
+        "导热垫设置在液冷板与电芯之间，厚度为1 mm，硬度为邵氏A 30。"
+    ]
+
+
+def test_long_candidates_are_flagged_and_empty_input_reported() -> None:
+    long_text = "导热垫" * (LONG_FEATURE_CHARS // 3 + 1)
+    assert any("P1" in w and "split during review" in w for w in split_description(long_text).warnings)
+    assert split_description("  \n ；。 ").warnings == ("no candidate features found",)
diff --git a/docs/adr/0010-product-features.md b/docs/adr/0010-product-features.md
new file mode 100644
index 0000000..0a9a3b2
--- /dev/null
+++ b/docs/adr/0010-product-features.md
@@ -0,0 +1,53 @@
+# ADR 0010: 产品技术特征——自由文本录入、规则拆分、人工确认
+
+- Status: Accepted（产品负责人于 2026-10-04 选定：自由文本，拆分成特征后人工确认）
+- Date: 2026-10-04
+
+## Context
+
+FTO 比对（ADR 0008）把权利要求的每个特征与“产品”逐项对照，因此需要产品的技术特征。产品负责人决定：
+用户用自由文本描述产品，系统拆分为候选特征，由人确认后才用于比对。
+
+## Decision
+
+### 1. 不复用现有 `feature_set_versions`，另建产品特征表
+
+ADR 0004 曾计划复用特征建模模块。核对代码后决定只复用其**流程模式**（草稿 → 确认 → 修订），不复用其表：
+
+- `feature_set_versions` 被比对、报告、检索三个服务按“本案最新已确认特征集”读取，三者都属于已冻结的可专利性流程。
+  产品特征若写入同一张表，会被这些流程当作交底书特征读取；避免这一点需要改动冻结代码。
+- 该表要求关联解析后的文档（`document_version_id` 非空），且特征类型是可专利性概念（前序/特征部分），与产品描述不符。
+
+### 2. 拆分：确定性规则，不使用大模型
+
+产品描述是客户保密信息，ADR 0005 规定的敏感度分级路由尚未实现，因此拆分**不调用任何大模型**，只用确定性规则：
+
+- 按换行拆分；行首的列表标记（`1.`、`1、`、`(1)`、`（1）`、`一、`、`-`、`•` 等）作为单元边界，不计入特征文本；
+- 行内按 `。；;！？` 以及英文句末（`.`/`!`/`?` 后接空白与新句首）拆分；常见缩写（e.g.、i.e.、etc.、Fig.、No.）不视为句末；
+  小数（如 `0.5 mm`）不会被拆开；中文逗号不拆分（同一技术特征常含多个分句）；
+- 每个候选特征保存其在原文中的起止位置；过长的候选特征（超过 120 个字符）提示人工拆分。
+
+规则拆分只给出候选，**不保证**与技术上的特征划分一致，因此必须经人工确认。
+
+### 3. 存储（机构数据，强制 RLS）
+
+产品描述属于客户的保密数据，与公开专利数据不同：三张表都带 `organization_id`、`case_id`，启用 FORCE RLS（ADR 0006 的全局表模式不适用）。
+
+| 表 | 内容 | 可变性 |
+|---|---|---|
+| `product_descriptions` | 产品描述全文的每个版本：文本、SHA-256、创建人 | 只追加（运行时角色仅 SELECT/INSERT） |
+| `product_feature_sets` | 某版描述的特征集：版本号、状态（`draft`/`confirmed`）、拆分规则版本、确认人与时间 | 草稿可改；确认后不可改、不可删 |
+| `product_features` | 特征：编号、文本、原文位置、来源（`split`/`manual`/`edited`）、顺序 | 所属特征集确认后不可增删改 |
+
+“确认后不可改”由数据库触发器强制（沿用 0001 迁移中的触发器模式），不只依赖应用代码：这是 AGENTS.md 要求的“审批后不可变”边界。
+修改已确认的特征集只能新建修订版本。
+
+### 4. 本期范围
+
+纯函数拆分器与模型、迁移、服务层（创建描述版本、生成草稿、编辑、拆分、合并、确认、修订）与测试。
+API 与界面在下一步。
+
+## Consequences
+
+- 新增 `modules/fto/product.py`（拆分器与模型）、迁移 `0021_product_features`、产品特征服务与测试。
+- 测试文本为**合成文本**（明确标注），不使用任何真实客户资料。
diff --git a/modules/fto/product.py b/modules/fto/product.py
new file mode 100644
index 0000000..b4e51d4
--- /dev/null
+++ b/modules/fto/product.py
@@ -0,0 +1,90 @@
+"""Product description → candidate technical features (ADR 0010).
+
+Deterministic rules only: product descriptions are customer-confidential and
+no sensitivity-routed model exists yet, so nothing here calls an LLM. The
+output is a set of *candidates* with exact spans in the original text; a person
+confirms, edits, splits or merges them before any comparison.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Literal
+
+SPLITTER_VERSION = "product-splitter/1"
+LONG_FEATURE_CHARS = 120  # review hint only: long candidates usually bundle several features
+
+FeatureOrigin = Literal["split", "manual", "edited"]
+
+# No "^": pattern.match(text, pos) already anchors at pos, while "^" would only match at index 0.
+# "-" and "*" count as bullets only before whitespace, so "-5 °C" keeps its sign.
+_LIST_MARKER = re.compile(
+    r"[ \t]*(?:-(?=\s)|\*(?=\s)|[•·▪]|\(?\d{1,3}[.)、](?!\d)|（\d{1,3}）|\(\d{1,3}\)|[一二三四五六七八九十]+、|[a-zA-Z][.)](?=\s))[ \t]*"
+)
+# Clause/sentence ends: Chinese and ASCII semicolons, Chinese full stop, ! and ?.
+_HARD_END = re.compile(r"[。；;！？!?]")
+# English sentence end: '.' then whitespace then an uppercase letter or CJK character.
+_EN_SENTENCE_END = re.compile(r"\.(?=\s+[A-Z一-鿿])")
+_ABBREVIATIONS = ("e.g.", "i.e.", "etc.", "fig.", "no.", "approx.", "vs.", "mr.", "dr.")
+
+
+@dataclass(frozen=True)
+class ProductFeature:
+    feature_id: str
+    text: str
+    spans: tuple[tuple[int, int], ...]  # into the description text; empty for manual features
+    origin: FeatureOrigin = "split"
+
+
+@dataclass(frozen=True)
+class SplitResult:
+    features: tuple[ProductFeature, ...]
+    warnings: tuple[str, ...]
+    splitter_version: str = SPLITTER_VERSION
+
+
+def _is_abbreviation(text: str, dot_index: int) -> bool:
+    window = text[max(0, dot_index - 7) : dot_index + 1].lower()
+    return any(window.endswith(a) for a in _ABBREVIATIONS)
+
+
+def _cut_points(text: str, start: int, end: int) -> list[int]:
+    cuts = []
+    for match in _HARD_END.finditer(text, start, end):
+        cuts.append(match.end())
+    for match in _EN_SENTENCE_END.finditer(text, start, end):
+        if not _is_abbreviation(text, match.start()):
+            cuts.append(match.end())
+    return sorted(set(cuts))
+
+
+def split_description(text: str) -> SplitResult:
+    features: list[ProductFeature] = []
+    warnings: list[str] = []
+    position = 0
+    for line in text.splitlines(keepends=True):
+        line_start, line_end = position, position + len(line)
+        position = line_end
+        marker = _LIST_MARKER.match(text, line_start, line_end)
+        content_start = marker.end() if marker else line_start
+        bounds = [content_start, *_cut_points(text, content_start, line_end), line_end]
+        for start, end in zip(bounds, bounds[1:]):
+            piece = text[start:end]
+            left = len(piece) - len(piece.lstrip())
+            right = len(piece.rstrip())
+            if right <= left:
+                continue
+            span = (start + left, start + right)
+            body = text[span[0] : span[1]]
+            if not re.search(r"\w", body):  # punctuation-only fragments carry no feature
+                continue
+            features.append(ProductFeature(feature_id=f"P{len(features) + 1}", text=body, spans=(span,)))
+    for feature in features:
+        if len(feature.text) > LONG_FEATURE_CHARS:
+            warnings.append(
+                f"{feature.feature_id} is {len(feature.text)} characters: may contain several features; split during review"
+            )
+    if not features:
+        warnings.append("no candidate features found")
+    return SplitResult(tuple(features), tuple(warnings))