diff --git a/tests/test_library.py b/tests/test_library.py
index f092fa769..188cd784d 100644
--- a/tests/test_library.py
+++ b/tests/test_library.py
@@ -11,6 +11,7 @@
 from httpx import ASGITransport, AsyncClient
 from tinyagentos.library_pipeline import (
     FileProcessor,
+    HeavyDownloadProcessor,
     ImageProcessor,
     PdfProcessor,
     TextProcessor,
@@ -1086,6 +1087,405 @@ async def test_reprocess_while_processing_returns_409(self, client, app):
         assert resp.status_code == 409
 
 
+# ---------------------------------------------------------------------------
+# P3: LibraryStore — rules + storage accounting
+# ---------------------------------------------------------------------------
+
+
+class TestLibraryStoreRules:
+    @pytest.mark.asyncio
+    async def test_create_and_list_rules(self, lib_store):
+        rid = await lib_store.create_rule(
+            source_pattern="*.youtube.com/*",
+            quality="1080",
+            auto_download=True,
+        )
+        assert rid
+
+        rules = await lib_store.list_rules()
+        assert len(rules) == 1
+        assert rules[0]["source_pattern"] == "*.youtube.com/*"
+        assert rules[0]["quality"] == "1080"
+        assert rules[0]["auto_download"] == 1
+
+    @pytest.mark.asyncio
+    async def test_get_rule(self, lib_store):
+        rid = await lib_store.create_rule(source_pattern="*.example.com/*")
+        rule = await lib_store.get_rule(rid)
+        assert rule is not None
+        assert rule["source_pattern"] == "*.example.com/*"
+
+        assert await lib_store.get_rule("nonexistent") is None
+
+    @pytest.mark.asyncio
+    async def test_delete_rule(self, lib_store):
+        rid = await lib_store.create_rule(source_pattern="*.youtube.com/*")
+        await lib_store.delete_rule(rid)
+        assert len(await lib_store.list_rules()) == 0
+
+    @pytest.mark.asyncio
+    async def test_match_rules(self, lib_store):
+        await lib_store.create_rule(
+            source_pattern="*youtube.com/*", quality="720", auto_download=True,
+        )
+        await lib_store.create_rule(
+            source_pattern="*vimeo.com/*", quality="1080", auto_download=False,
+        )
+
+        matched = await lib_store.match_rules("https://www.youtube.com/watch?v=abc")
+        assert len(matched) == 1
+        assert matched[0]["quality"] == "720"
+
+        matched_vimeo = await lib_store.match_rules("https://vimeo.com/12345")
+        assert len(matched_vimeo) == 1
+
+        matched_none = await lib_store.match_rules("https://example.com")
+        assert len(matched_none) == 0
+
+    @pytest.mark.asyncio
+    async def test_match_rules_disabled_ignored(self, lib_store):
+        rid = await lib_store.create_rule(
+            source_pattern="*example.com/*",
+            enabled=False,
+        )
+        matched = await lib_store.match_rules("https://example.com/page")
+        assert len(matched) == 0
+
+    @pytest.mark.asyncio
+    async def test_storage_summary(self, lib_store):
+        await lib_store.create_item(kind="text", title="a.txt", size_bytes=100)
+        await lib_store.create_item(kind="pdf", title="b.pdf", size_bytes=500)
+        await lib_store.create_item(kind="url:youtube", title="c", size_bytes=0)
+
+        summary = await lib_store.get_storage_summary()
+        assert summary["total_count"] == 3
+        assert summary["total_bytes"] == 600
+        assert "text" in summary["by_kind"]
+        assert "pdf" in summary["by_kind"]
+
+    @pytest.mark.asyncio
+    async def test_storage_summary_empty(self, lib_store):
+        summary = await lib_store.get_storage_summary()
+        assert summary["total_count"] == 0
+        assert summary["total_bytes"] == 0
+
+
+# ---------------------------------------------------------------------------
+# P3: HeavyDownloadProcessor
+# ---------------------------------------------------------------------------
+
+
+class TestHeavyDownloadProcessor:
+    @pytest.mark.asyncio
+    async def test_process_heavy_download(self, lib_store, storage_dir):
+        item_id = await lib_store.create_item(
+            kind="url:youtube",
+            source_url="https://youtube.com/watch?v=heavy-test",
+        )
+        await lib_store.update_item(item_id, quality="480")
+        item = await lib_store.get_item(item_id)
+
+        proc = HeavyDownloadProcessor(lib_store, storage_dir)
+
+        from unittest.mock import patch
+
+        mock_path = str(storage_dir / "downloads" / "test123.mp4")
+        storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True)
+        storage_dir.joinpath("downloads", "test123.mp4").write_text("fake video data")
+
+        with patch(
+            "tinyagentos.knowledge_fetchers.youtube.download_video",
+            _async_return(mock_path),
+        ):
+            artifacts = await proc.process(item)
+
+        assert len(artifacts) == 1
+        assert artifacts[0]["kind"] == "download"
+        assert artifacts[0]["meta"]["quality"] == "480"
+
+        updated = await lib_store.get_item(item_id)
+        assert updated["download_path"] == mock_path
+        assert updated["download_bytes"] > 0
+
+    @pytest.mark.asyncio
+    async def test_process_heavy_download_non_youtube(self, lib_store, storage_dir):
+        item_id = await lib_store.create_item(
+            kind="url:web",
+            source_url="https://example.com",
+        )
+        item = await lib_store.get_item(item_id)
+
+        proc = HeavyDownloadProcessor(lib_store, storage_dir)
+        artifacts = await proc.process(item)
+        assert artifacts == []
+
+    @pytest.mark.asyncio
+    async def test_process_heavy_download_no_source(self, lib_store, storage_dir):
+        item_id = await lib_store.create_item(
+            kind="url:youtube", source_url="",
+        )
+        item = await lib_store.get_item(item_id)
+
+        proc = HeavyDownloadProcessor(lib_store, storage_dir)
+        artifacts = await proc.process(item)
+        assert artifacts == []
+
+    @pytest.mark.asyncio
+    async def test_process_heavy_download_invalid_quality(self, lib_store, storage_dir):
+        """Invalid quality should fall back to 720."""
+        item_id = await lib_store.create_item(
+            kind="url:youtube",
+            source_url="https://youtube.com/watch?v=qtest",
+        )
+        await lib_store.update_item(item_id, quality="9999")
+        item = await lib_store.get_item(item_id)
+
+        proc = HeavyDownloadProcessor(lib_store, storage_dir)
+
+        from unittest.mock import patch
+
+        mock_path = str(storage_dir / "downloads" / "test.mp4")
+        storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True)
+        storage_dir.joinpath("downloads", "test.mp4").write_text("data")
+
+        with patch(
+            "tinyagentos.knowledge_fetchers.youtube.download_video",
+            _async_return(mock_path),
+        ):
+            artifacts = await proc.process(item)
+
+        # Quality should have been normalized to 720
+        assert artifacts[0]["meta"]["quality"] == "720"
+
+
+# ---------------------------------------------------------------------------
+# P3: run_heavy_pipeline
+# ---------------------------------------------------------------------------
+
+
+class TestRunHeavyPipeline:
+    @pytest.mark.asyncio
+    async def test_run_heavy_pipeline_rule_quality(self, lib_store, storage_dir):
+        """When a matching rule exists, its quality is used."""
+        await lib_store.create_rule(
+            source_pattern="*youtube.com/*",
+            quality="480",
+            auto_download=False,
+        )
+
+        item_id = await lib_store.create_item(
+            kind="url:youtube",
+            source_url="https://youtube.com/watch?v=rule-test",
+        )
+        storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True)
+
+        mock_path = str(storage_dir / "downloads" / "test.mp4")
+        storage_dir.joinpath("downloads", "test.mp4").write_text("fake data")
+
+        from unittest.mock import patch
+        from tinyagentos.library_pipeline import run_heavy_pipeline
+
+        with patch(
+            "tinyagentos.knowledge_fetchers.youtube.download_video",
+            _async_return(mock_path),
+        ):
+            result = await run_heavy_pipeline(
+                lib_store, item_id, storage_dir, quality="",
+            )
+
+        assert result is not None
+        assert result["quality"] == "480"  # From rule
+
+    @pytest.mark.asyncio
+    async def test_run_heavy_pipeline_explicit_quality(self, lib_store, storage_dir):
+        """Explicit quality parameter overrides rule quality."""
+        await lib_store.create_rule(
+            source_pattern="*youtube.com/*",
+            quality="480",
+        )
+
+        item_id = await lib_store.create_item(
+            kind="url:youtube",
+            source_url="https://youtube.com/watch?v=explicit-test",
+        )
+        storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True)
+
+        mock_path = str(storage_dir / "downloads" / "test.mp4")
+        storage_dir.joinpath("downloads", "test.mp4").write_text("fake data")
+
+        from unittest.mock import patch
+        from tinyagentos.library_pipeline import run_heavy_pipeline
+
+        with patch(
+            "tinyagentos.knowledge_fetchers.youtube.download_video",
+            _async_return(mock_path),
+        ):
+            result = await run_heavy_pipeline(
+                lib_store, item_id, storage_dir, quality="1080",
+            )
+
+        assert result["quality"] == "1080"  # Explicit wins
+
+    @pytest.mark.asyncio
+    async def test_run_heavy_pipeline_non_youtube(self, lib_store, storage_dir):
+        """run_heavy_pipeline returns None for non-YouTube items."""
+        item_id = await lib_store.create_item(
+            kind="url:web",
+            source_url="https://example.com/page",
+        )
+
+        from tinyagentos.library_pipeline import run_heavy_pipeline
+        result = await run_heavy_pipeline(
+            lib_store, item_id, storage_dir,
+        )
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_run_heavy_pipeline_creates_job(self, lib_store, storage_dir):
+        """Heavy pipeline creates a job entry."""
+        item_id = await lib_store.create_item(
+            kind="url:youtube",
+            source_url="https://youtube.com/watch?v=job-test",
+        )
+        storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True)
+
+        mock_path = str(storage_dir / "downloads" / "test.mp4")
+        storage_dir.joinpath("downloads", "test.mp4").write_text("fake data")
+
+        from unittest.mock import patch
+        from tinyagentos.library_pipeline import run_heavy_pipeline
+
+        with patch(
+            "tinyagentos.knowledge_fetchers.youtube.download_video",
+            _async_return(mock_path),
+        ):
+            await run_heavy_pipeline(lib_store, item_id, storage_dir, quality="720")
+
+        jobs = await lib_store.get_item_jobs(item_id)
+        heavy_jobs = [j for j in jobs if j["stage"] == "heavy_download"]
+        assert len(heavy_jobs) >= 1
+
+
+# ---------------------------------------------------------------------------
+# P3: Library routes — download, rules, usage
+# ---------------------------------------------------------------------------
+
+
+class TestLibraryRoutesP3:
+    @pytest.mark.asyncio
+    async def test_trigger_download(self, client):
+        """POST /api/library/items/{id}/download triggers heavy download."""
+        # Ingest a YouTube URL first
+        resp = await client.post(
+            "/api/library/ingest", data={"url": "https://youtube.com/watch?v=dl-test"}
+        )
+        assert resp.status_code == 202
+        item_id = resp.json()["item_id"]
+
+        # Allow pipeline to finish
+        import asyncio as _asyncio
+        await _asyncio.sleep(0.5)
+
+        resp = await client.post(
+            f"/api/library/items/{item_id}/download",
+            data={"quality": "480"},
+        )
+        assert resp.status_code == 202
+        data = resp.json()
+        assert data["status"] == "downloading"
+        assert data["quality"] == "480"
+
+    @pytest.mark.asyncio
+    async def test_trigger_download_nonexistent(self, client):
+        resp = await client.post("/api/library/items/nonexistent/download")
+        assert resp.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_trigger_download_non_youtube(self, client):
+        """Download endpoint rejects non-YouTube items."""
+        resp = await client.post(
+            "/api/library/ingest", data={"url": "https://example.com/page"}
+        )
+        assert resp.status_code == 202
+        item_id = resp.json()["item_id"]
+
+        import asyncio as _asyncio
+        await _asyncio.sleep(0.5)
+
+        resp = await client.post(f"/api/library/items/{item_id}/download")
+        assert resp.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_download_status(self, client):
+        resp = await client.post(
+            "/api/library/ingest", data={"url": "https://youtube.com/watch?v=status"}
+        )
+        item_id = resp.json()["item_id"]
+
+        import asyncio as _asyncio
+        await _asyncio.sleep(0.5)
+
+        resp = await client.get(f"/api/library/items/{item_id}/download/status")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["item_id"] == item_id
+        assert "downloaded" in data
+
+    @pytest.mark.asyncio
+    async def test_create_rule(self, client):
+        resp = await client.post(
+            "/api/library/rules",
+            data={"source_pattern": "*.youtube.com/*", "quality": "1080", "auto_download": "true"},
+        )
+        assert resp.status_code == 201
+        data = resp.json()
+        assert data["rule_id"]
+        assert data["source_pattern"] == "*.youtube.com/*"
+
+    @pytest.mark.asyncio
+    async def test_list_rules(self, client):
+        # Create a rule first
+        await client.post(
+            "/api/library/rules", data={"source_pattern": "*.example.com/*"}
+        )
+        resp = await client.get("/api/library/rules")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "rules" in data
+        assert data["count"] >= 1
+
+    @pytest.mark.asyncio
+    async def test_delete_rule(self, client):
+        resp = await client.post(
+            "/api/library/rules", data={"source_pattern": "*.test.com/*"}
+        )
+        rule_id = resp.json()["rule_id"]
+
+        resp = await client.delete(f"/api/library/rules/{rule_id}")
+        assert resp.status_code == 200
+        assert resp.json()["status"] == "deleted"
+
+        # Verify it's gone
+        resp = await client.get("/api/library/rules")
+        data = resp.json()
+        rule_ids = [r["id"] for r in data["rules"]]
+        assert rule_id not in rule_ids
+
+    @pytest.mark.asyncio
+    async def test_delete_nonexistent_rule(self, client):
+        resp = await client.delete("/api/library/rules/nonexistent")
+        assert resp.status_code == 404
+
+    @pytest.mark.asyncio
+    async def test_storage_usage(self, client):
+        resp = await client.get("/api/library/usage")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "total_count" in data
+        assert "total_bytes" in data
+        assert "by_kind" in data
+
+
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py
index 5435037d2..14ede56d6 100644
--- a/tinyagentos/library_pipeline.py
+++ b/tinyagentos/library_pipeline.py
@@ -650,6 +650,184 @@ def get_processor(kind: str, store: LibraryStore,
     return cls(store, storage_dir)
 
 
+# ---------------------------------------------------------------------------
+# Heavy tier — opt-in media download
+# ---------------------------------------------------------------------------
+
+
+class HeavyDownloadProcessor(Processor):
+    """Downloads media for items that have opted into the heavy tier.
+
+    Currently supports url:youtube items via yt-dlp download_video.
+    Respects per-item quality preference and per-source rules.
+    """
+
+    _VALID_QUALITIES = frozenset({"360", "480", "720", "1080", "best"})
+
+    async def process(self, item: dict) -> list[dict]:
+        item_id = item["id"]
+        source_url = item.get("source_url", "")
+        artifacts: list[dict] = []
+
+        if not source_url:
+            return artifacts
+
+        kind = item.get("kind", "")
+        if kind != "url:youtube":
+            return artifacts
+
+        quality = item.get("quality", "") or "720"
+        if quality not in self._VALID_QUALITIES:
+            quality = "720"
+
+        try:
+            from tinyagentos.knowledge_fetchers.youtube import download_video
+        except ImportError:
+            logger.warning("yt-dlp not available for heavy download")
+            return artifacts
+
+        download_dir = self.storage_dir / "downloads"
+        download_dir.mkdir(parents=True, exist_ok=True)
+
+        path = await download_video(source_url, quality=quality, output_dir=download_dir)
+
+        if not path:
+            msg = f"Heavy download failed for {source_url!r}: yt-dlp returned no output path"
+            logger.warning(msg)
+            await self.store.update_item(
+                item_id,
+                meta_json={
+                    **json.loads(item.get("meta_json", "{}")),
+                    "download_error": msg,
+                },
+            )
+            return artifacts
+
+        p = Path(path)
+        if not p.exists():
+            # yt-dlp skips printing a Destination line when the file already
+            # exists, so fall back to locating it on disk.  Scope the search to
+            # THIS item's video id so concurrent downloads of different videos
+            # cannot cross-attribute each other's files.
+            stored_meta = json.loads(item.get("meta_json", "{}"))
+            video_id = stored_meta.get("video_id", "")
+            candidates: list[Path] = []
+            if video_id:
+                candidates = sorted(
+                    (c for c in download_dir.glob(f"{video_id}*") if c.is_file()),
+                    key=lambda x: x.stat().st_mtime,
+                    reverse=True,
+                )
+            if candidates:
+                p = candidates[0]
+            else:
+                await self.store.update_item(
+                    item_id,
+                    meta_json={
+                        **json.loads(item.get("meta_json", "{}")),
+                        "download_error": "Downloaded file not found on disk",
+                    },
+                )
+                return artifacts
+
+        size_bytes = p.stat().st_size
+        await self.store.update_item(
+            item_id,
+            download_path=str(p),
+            download_bytes=size_bytes,
+            bytes=size_bytes,
+            downloaded_at=time.time(),
+        )
+
+        download_meta: dict = {
+            "path": str(p),
+            "bytes": size_bytes,
+            "quality": quality,
+            "format": p.suffix.lstrip("."),
+        }
+        await self.store.add_artifact(
+            item_id, kind="download", path=str(p), meta=download_meta,
+        )
+        artifacts.append({
+            "kind": "download", "path": str(p), "meta": download_meta,
+        })
+
+        return artifacts
+
+
+async def run_heavy_pipeline(
+    store: LibraryStore,
+    item_id: str,
+    storage_dir: Path,
+    quality: str = "",
+) -> dict | None:
+    """Run the heavy-tier download pipeline for one item.
+
+    Checks per-source rules for auto_download settings, respects per-item
+    quality override, and downloads the media via yt-dlp.
+
+    Returns download metadata dict on success, None if skipped or failed.
+    """
+    item = await store.get_item(item_id)
+    if not item:
+        return None
+
+    kind = item.get("kind", "")
+    source_url = item.get("source_url", "")
+
+    # Only YouTube items are supported for heavy download currently
+    if kind != "url:youtube" or not source_url:
+        return None
+
+    # Check for matching rules (apply first matching rule's quality if
+    # no explicit quality was provided)
+    if not quality:
+        rules = await store.match_rules(source_url)
+        if rules:
+            quality = rules[0].get("quality", "") or "720"
+
+    # Fallback to item's quality field, then default 720
+    if not quality:
+        quality = item.get("quality", "") or "720"
+
+    # Create a job entry
+    await store.create_job(item_id, "heavy_download")
+
+    try:
+        proc = HeavyDownloadProcessor(store, storage_dir)
+        # Override the item's quality for this run
+        item_with_quality = dict(item, quality=quality)
+        artifacts = await proc.process(item_with_quality)
+
+        if artifacts:
+            await store.update_job(
+                (await store.get_item_jobs(item_id))[-1]["id"],
+                state="done",
+            )
+            return artifacts[0].get("meta", {})
+        else:
+            await store.update_job(
+                (await store.get_item_jobs(item_id))[-1]["id"],
+                state="error",
+                error="Download produced no artifacts",
+            )
+            return None
+    except Exception:
+        logger.exception("Heavy pipeline failed for item %s", item_id)
+        # Heavy download is OPTIONAL — the item is already 'ready' from the
+        # cheap-tier ingest.  Do not flip it to 'error'; surface the failure on
+        # the heavy_download job instead (queryable via /download/status).
+        try:
+            jobs = await store.get_item_jobs(item_id)
+            if jobs:
+                await store.update_job(
+                    jobs[-1]["id"], state="error", error="Heavy download failed"
+                )
+        except Exception:
+            logger.exception("Failed to record heavy download error for %s", item_id)
+        return None
+
+
 # ---------------------------------------------------------------------------
 # Pipeline runner
 # ---------------------------------------------------------------------------
diff --git a/tinyagentos/library_store.py b/tinyagentos/library_store.py
index 6d7b1d11c..2500e5044 100644
--- a/tinyagentos/library_store.py
+++ b/tinyagentos/library_store.py
@@ -59,6 +59,16 @@
 );
 CREATE INDEX IF NOT EXISTS idx_lj_item ON library_jobs(item_id);
 CREATE INDEX IF NOT EXISTS idx_lj_state ON library_jobs(state);
+
+CREATE TABLE IF NOT EXISTS library_rules (
+    id TEXT PRIMARY KEY,
+    source_pattern TEXT NOT NULL,
+    quality TEXT NOT NULL DEFAULT '720',
+    auto_download INTEGER NOT NULL DEFAULT 0,
+    enabled INTEGER NOT NULL DEFAULT 1,
+    created_at REAL NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_lr_pattern ON library_rules(source_pattern);
 """
 
 _VALID_STATUSES = frozenset({"pending", "processing", "ready", "error"})
@@ -68,10 +78,28 @@ class LibraryStore(BaseStore):
     SCHEMA = LIBRARY_SCHEMA
 
     async def _post_init(self) -> None:
-        """Enable foreign key enforcement for cascade deletes."""
+        """Enable foreign key enforcement and apply schema migrations."""
         await self._db.execute("PRAGMA foreign_keys = ON")
         await self._db.commit()
 
+        # P3 migration: add quality, auto_download, downloaded_at columns
+        # if they don't exist yet (existing databases from P1/P2).
+        cols = await self._db.execute_fetchall("PRAGMA table_info(library_items)")
+        col_names = {row[1] for row in cols}
+        for col, col_def in [
+            ("quality", "TEXT NOT NULL DEFAULT ''"),
+            ("auto_download", "INTEGER NOT NULL DEFAULT 0"),
+            ("downloaded_at", "REAL"),
+            ("download_path", "TEXT NOT NULL DEFAULT ''"),
+            ("download_bytes", "INTEGER NOT NULL DEFAULT 0"),
+        ]:
+            if col not in col_names:
+                await self._db.execute(
+                    f"ALTER TABLE library_items ADD COLUMN {col} {col_def}"
+                )
+        if col_names:
+            await self._db.commit()
+
     # -- items ------------------------------------------------------------
 
     async def create_item(
@@ -140,7 +168,9 @@ async def list_items(
 
     async def update_item(self, item_id: str, **kwargs) -> None:
         allowed = {"kind", "source_url", "title", "status", "storage_path",
-                    "bytes", "meta_json", "updated_at"}
+                    "bytes", "meta_json", "updated_at",
+                    "quality", "auto_download", "download_path",
+                    "download_bytes", "downloaded_at"}
         fields = [(k, v) for k, v in kwargs.items() if k in allowed]
         if not fields:
             return
@@ -273,3 +303,83 @@ async def update_job(self, job_id: str, **kwargs) -> None:
             f"UPDATE library_jobs SET {set_clause} WHERE id = ?", values
         )
         await self._db.commit()
+
+    # -- source rules -----------------------------------------------------
+
+    async def create_rule(
+        self,
+        source_pattern: str,
+        quality: str = "720",
+        auto_download: bool = False,
+        enabled: bool = True,
+    ) -> str:
+        rule_id = uuid.uuid4().hex[:16]
+        now = time.time()
+        await self._db.execute(
+            """INSERT INTO library_rules
+               (id, source_pattern, quality, auto_download, enabled, created_at)
+               VALUES (?, ?, ?, ?, ?, ?)""",
+            (rule_id, source_pattern, quality, int(auto_download), int(enabled), now),
+        )
+        await self._db.commit()
+        return rule_id
+
+    async def list_rules(self) -> list[dict]:
+        self._db.row_factory = aiosqlite.Row
+        async with self._db.execute(
+            "SELECT * FROM library_rules ORDER BY created_at DESC"
+        ) as cursor:
+            rows = await cursor.fetchall()
+        return [dict(r) for r in rows]
+
+    async def get_rule(self, rule_id: str) -> dict | None:
+        self._db.row_factory = aiosqlite.Row
+        async with self._db.execute(
+            "SELECT * FROM library_rules WHERE id = ?", (rule_id,)
+        ) as cursor:
+            row = await cursor.fetchone()
+            return dict(row) if row else None
+
+    async def delete_rule(self, rule_id: str) -> None:
+        await self._db.execute(
+            "DELETE FROM library_rules WHERE id = ?", (rule_id,)
+        )
+        await self._db.commit()
+
+    async def match_rules(self, source_url: str) -> list[dict]:
+        """Return all enabled rules whose source_pattern matches source_url."""
+        self._db.row_factory = aiosqlite.Row
+        async with self._db.execute(
+            "SELECT * FROM library_rules WHERE enabled = 1 ORDER BY created_at"
+        ) as cursor:
+            rows = await cursor.fetchall()
+        rules = [dict(r) for r in rows]
+        import fnmatch
+        return [
+            r for r in rules
+            if fnmatch.fnmatch(source_url.lower(), r["source_pattern"].lower())
+        ]
+
+    # -- storage accounting -----------------------------------------------
+
+    async def get_storage_summary(self) -> dict:
+        """Return total bytes, item count, and per-kind breakdown."""
+        self._db.row_factory = aiosqlite.Row
+        rows = await self._db.execute_fetchall(
+            "SELECT kind, COUNT(*) as count, "
+            "COALESCE(SUM(bytes), 0) as total_bytes "
+            "FROM library_items GROUP BY kind"
+        )
+        by_kind = {row["kind"]: {"count": row["count"], "bytes": row["total_bytes"]} for row in rows}
+
+        total = await self._db.execute_fetchall(
+            "SELECT COUNT(*) as total_count, COALESCE(SUM(bytes), 0) as total_bytes "
+            "FROM library_items"
+        )
+        if total:
+            return {
+                "total_count": total[0]["total_count"],
+                "total_bytes": total[0]["total_bytes"],
+                "by_kind": by_kind,
+            }
+        return {"total_count": 0, "total_bytes": 0, "by_kind": {}}
diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py
index 745eb913e..447d05819 100644
--- a/tinyagentos/routes/library.py
+++ b/tinyagentos/routes/library.py
@@ -46,6 +46,7 @@ def _on_done(t: asyncio.Task) -> None:
 
 # ---------------------------------------------------------------------------
 
+
 def _library_dir_from_app(app) -> Path:
     """Return the library storage directory, creating it if needed."""
     data_dir = getattr(app.state, "data_dir", None)
@@ -187,6 +188,27 @@ async def _ingest_task(app, item_id: str, store, storage_dir: Path) -> None:
         await store.update_item_status(item_id, "error")
         return
 
+    # Auto-download: check matching rules with auto_download=True
+    try:
+        item = await store.get_item(item_id)
+        if item and item.get("source_url"):
+            rules = await store.match_rules(item["source_url"])
+            for rule in rules:
+                if rule.get("auto_download"):
+                    from tinyagentos.library_pipeline import run_heavy_pipeline
+
+                    quality = rule.get("quality", "") or "720"
+                    logger.info(
+                        "Auto-download triggered for item %s by rule %s (quality=%s)",
+                        item_id, rule["id"], quality,
+                    )
+                    await run_heavy_pipeline(
+                        store, item_id, storage_dir, quality=quality,
+                    )
+                    break  # First matching auto-download rule wins
+    except Exception:
+        logger.exception("Auto-download check failed for item %s", item_id)
+
     # Collections handoff after successful pipeline
     try:
         collections_dir = storage_dir.parent / "collections"
@@ -371,3 +393,143 @@ async def reprocess_item(request: Request, item_id: str):
         )
 
     return JSONResponse({"item_id": item_id, "status": "reprocessing"}, status_code=202)
+
+
+# ---------------------------------------------------------------------------
+# Heavy tier: download
+# ---------------------------------------------------------------------------
+
+
+@router.post("/api/library/items/{item_id}/download")
+async def trigger_download(
+    request: Request,
+    item_id: str,
+    quality: str | None = Form(None),
+):
+    """Trigger heavy-tier media download for an item.
+
+    Accepts optional ``quality`` form field (360, 480, 720, 1080, best).
+    Runs the download asynchronously in a background task.
+    """
+    store = await _get_library_store(request)
+    item = await store.get_item(item_id)
+    if not item:
+        return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404)
+
+    if item["kind"] != "url:youtube":
+        return JSONResponse(
+            {"error": "Heavy download only supports url:youtube items"},
+            status_code=400,
+        )
+
+    storage_dir = _library_dir(request)
+    quality_val = quality or item.get("quality", "") or "720"
+
+    task_set = getattr(request.app.state, "_background_tasks", None)
+    coro = _heavy_download_task(request.app, item_id, store, storage_dir, quality_val)
+    if task_set is None:
+        _track_background_task(coro)
+    else:
+        _create_supervised_task(coro, task_set)
+
+    return JSONResponse(
+        {"item_id": item_id, "status": "downloading", "quality": quality_val},
+        status_code=202,
+    )
+
+
+async def _heavy_download_task(
+    app, item_id: str, store, storage_dir: Path, quality: str
+) -> None:
+    """Background task: run heavy download for an item."""
+    from tinyagentos.library_pipeline import run_heavy_pipeline
+
+    try:
+        result = await run_heavy_pipeline(store, item_id, storage_dir, quality=quality)
+        if result:
+            logger.info("Heavy download complete for item %s: %s", item_id, result)
+    except Exception:
+        logger.exception("Heavy download crashed for item %s", item_id)
+        # Heavy download is OPTIONAL — leave the item 'ready'.  The failure is
+        # surfaced through the heavy_download job state (run_heavy_pipeline
+        # records it) or via this log line.
+
+
+@router.get("/api/library/items/{item_id}/download/status")
+async def download_status(request: Request, item_id: str):
+    """Check heavy download status for an item."""
+    store = await _get_library_store(request)
+    item = await store.get_item(item_id)
+    if not item:
+        return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404)
+
+    jobs = await store.get_item_jobs(item_id)
+    heavy_jobs = [j for j in jobs if j.get("stage") == "heavy_download"]
+    download_path = item.get("download_path", "")
+    download_bytes = item.get("download_bytes", 0)
+
+    return {
+        "item_id": item_id,
+        "downloaded": bool(download_path),
+        "download_path": download_path,
+        "download_bytes": download_bytes,
+        "jobs": heavy_jobs,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Source rules
+# ---------------------------------------------------------------------------
+
+
+@router.post("/api/library/rules")
+async def create_rule(
+    request: Request,
+    source_pattern: str = Form(...),
+    quality: str | None = Form("720"),
+    auto_download: bool = Form(False),
+):
+    """Create a source rule for automatic heavy download triggers."""
+    store = await _get_library_store(request)
+    rule_id = await store.create_rule(
+        source_pattern=source_pattern,
+        quality=quality or "720",
+        auto_download=auto_download,
+    )
+    return JSONResponse(
+        {"rule_id": rule_id, "source_pattern": source_pattern, "status": "created"},
+        status_code=201,
+    )
+
+
+@router.get("/api/library/rules")
+async def list_rules(request: Request):
+    """List all source rules."""
+    store = await _get_library_store(request)
+    rules = await store.list_rules()
+    return {"rules": rules, "count": len(rules)}
+
+
+@router.delete("/api/library/rules/{rule_id}")
+async def delete_rule(request: Request, rule_id: str):
+    """Delete a source rule."""
+    store = await _get_library_store(request)
+    rule = await store.get_rule(rule_id)
+    if not rule:
+        return JSONResponse({"error": f"Rule {rule_id!r} not found"}, status_code=404)
+
+    await store.delete_rule(rule_id)
+    return {"status": "deleted", "rule_id": rule_id}
+
+
+# ---------------------------------------------------------------------------
+# Storage accounting
+# ---------------------------------------------------------------------------
+
+
+@router.get("/api/library/usage")
+async def storage_usage(request: Request):
+    """Return storage accounting summary for the library."""
+    store = await _get_library_store(request)
+    summary = await store.get_storage_summary()
+    return summary