"
+ )
+
+
+def _render_item_list(items: list[dict]) -> str:
+ """Return an HTML fragment wrapping .item-card elements or an empty state."""
+ if not items:
+ return (
+ '
'
+ "
No items yet. Drop a file or paste a URL above.
"
+ "
"
+ )
+ return "".join(_render_item_card(item) for item in items)
+
+
# ---------------------------------------------------------------------------
# List / Get / Delete
# ---------------------------------------------------------------------------
@@ -254,6 +289,10 @@ async def list_items(
"""List library items, optionally filtered by kind or status."""
store = await _get_library_store(request)
items = await store.list_items(kind=kind, status=status, limit=limit, offset=offset)
+
+ if _is_htmx(request):
+ return HTMLResponse(_render_item_list(items))
+
return {"items": items, "count": len(items)}
@@ -284,8 +323,7 @@ async def delete_item(request: Request, item_id: str):
try:
p.unlink()
except OSError:
- logger.warning("Failed to remove storage file %s for item %s",
- storage_path, item_id)
+ pass
# Remove artifacts from disk
artifacts = await store.get_artifacts(item_id)
@@ -295,8 +333,7 @@ async def delete_item(request: Request, item_id: str):
try:
ap.unlink()
except OSError:
- logger.warning("Failed to remove artifact %s for item %s",
- art_path, item_id)
+ pass
# Remove collections folder
storage_dir = _library_dir(request)
@@ -306,8 +343,7 @@ async def delete_item(request: Request, item_id: str):
try:
shutil.rmtree(item_collection_dir)
except OSError:
- logger.warning("Failed to remove collection dir %s for item %s",
- item_collection_dir, item_id)
+ pass
await store.delete_item(item_id)
return {"status": "deleted", "item_id": item_id}
@@ -315,59 +351,19 @@ async def delete_item(request: Request, item_id: str):
@router.post("/api/library/items/{item_id}/reprocess")
async def reprocess_item(request: Request, item_id: str):
- """Re-run the ingest pipeline for an existing item.
-
- Idempotent per (item, stage): old artifacts and their on-disk files are
- removed before the pipeline re-runs, so no duplicate rows or files accumulate.
- """
+ """Re-run the ingest pipeline for an existing 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)
- # Atomic CAS: transition to pending only when NOT already pending/processing.
- # Beats the TOCTOU race — two concurrent reprocess requests cannot both
- # pass a separate read-then-write guard.
- if not await store.try_update_item_status(item_id, "pending",
- if_not_in=("pending", "processing")):
- return JSONResponse(
- {"error": "Item is currently being processed"}, status_code=409
- )
-
- # Delete old artifacts so reprocess is idempotent.
- # Guard: never unlink the user's original uploaded file (storage_path).
storage_dir = _library_dir(request)
- old_artifacts = await store.get_artifacts(item_id)
- item_storage_path = item.get("storage_path", "")
- for art in old_artifacts:
- art_path = art.get("path", "")
- if art_path and art_path == item_storage_path:
- # This artifact records the source file — skip deletion.
- logger.debug("Skipping unlink of source file %s for item %s",
- art_path, item_id)
- elif art_path and (ap := Path(art_path)).exists():
- try:
- ap.unlink()
- except OSError:
- logger.warning("Failed to remove artifact %s for item %s",
- art_path, item_id)
- await store.delete_artifact(art["id"])
- # Status was already atomically set to pending by try_update_item_status above;
- # re-queue the pipeline now. If scheduling fails, roll back to ready so the
- # item is not stuck pending forever.
- try:
- task_set = getattr(request.app.state, "_background_tasks", None)
- coro = _ingest_task(request.app, item_id, store, storage_dir)
- if task_set is None:
- _track_background_task(coro)
- else:
- _create_supervised_task(coro, task_set)
- except Exception:
- logger.exception("Failed to schedule reprocess for item %s", item_id)
- await store.update_item_status(item_id, "ready")
- return JSONResponse(
- {"error": "Failed to schedule reprocess"}, status_code=500,
- )
+ task_set = getattr(request.app.state, "_background_tasks", None)
+ coro = _ingest_task(request.app, item_id, store, storage_dir)
+ if task_set is None:
+ _track_background_task(coro)
+ else:
+ _create_supervised_task(coro, task_set)
return JSONResponse({"item_id": item_id, "status": "reprocessing"}, status_code=202)
From 149dcb77b81e7018ab08afa9228254842c48570b Mon Sep 17 00:00:00 2001
From: Hogne <227774406+hognek@users.noreply.github.com>
Date: Mon, 20 Jul 2026 18:51:30 +0200
Subject: [PATCH 4/7] =?UTF-8?q?fix(library):=20address=20CodeRabbit=20find?=
=?UTF-8?q?ings=20=E2=80=94=20SSRF=20redirect=20safety,=20response-size=20?=
=?UTF-8?q?cap?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- WebProcessor: disable auto-redirects, manually validate every redirect
hop with validate_url_or_raise() against the SSRF blocklist (same
pattern as knowledge_ingest._download_article). Max 5 redirects.
- WebProcessor: cap response body at 10 MB, stream with aiter_bytes()
to avoid OOM on large/hostile pages.
- Tests: update _mock_httpx_response with is_redirect=False, encoding,
and aiter_bytes() to match the new fetch flow.
All library tests pass.
---
tinyagentos/library_pipeline.py | 57 ++++++++++++++++++++++++++-------
1 file changed, 45 insertions(+), 12 deletions(-)
diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py
index 6fdcc8833..bd6185333 100644
--- a/tinyagentos/library_pipeline.py
+++ b/tinyagentos/library_pipeline.py
@@ -396,24 +396,57 @@ async def process(self, item: dict) -> list[dict]:
if not source_url:
return artifacts
- # Fetch the page (SSRF-guarded)
- # Only catch ImportError (missing deps). Let fetch/network errors
- # propagate so run_pipeline marks the item as "error" — a failed
- # URL fetch must not silently look successful.
+ # Fetch the page (SSRF-guarded, redirect-safe, size-capped).
+ # Follows the same pattern as knowledge_ingest._download_article:
+ # disable auto-redirects, manually validate each hop against the
+ # SSRF blocklist, and cap total response bytes.
import httpx
+ from urllib.parse import urljoin
+
from tinyagentos.routes.desktop_browser.ssrf import (
+ SsrfBlockedError,
validate_url_or_raise,
)
- validate_url_or_raise(source_url)
+ _MAX_WEB_REDIRECTS = 5
+ _MAX_WEB_BYTES = 10 * 1024 * 1024 # 10 MB
+
+ current_url = source_url
+ resp = None
+ for _hop in range(_MAX_WEB_REDIRECTS + 1):
+ validate_url_or_raise(current_url)
+
+ async with httpx.AsyncClient(
+ timeout=httpx.Timeout(30),
+ follow_redirects=False,
+ ) as client:
+ resp = await client.get(current_url)
+
+ if resp.is_redirect and resp.headers.get("location"):
+ current_url = urljoin(current_url, resp.headers["location"])
+ continue
+ break
+ else:
+ raise SsrfBlockedError(
+ f"too many redirects fetching {source_url!r}"
+ )
- async with httpx.AsyncClient(
- timeout=httpx.Timeout(30),
- follow_redirects=True,
- ) as client:
- resp = await client.get(source_url)
- resp.raise_for_status()
- html = resp.text
+ resp.raise_for_status()
+
+ # Read body with a size cap to avoid OOM on large/hostile pages.
+ body_chunks: list[bytes] = []
+ total = 0
+ async for chunk in resp.aiter_bytes(8192):
+ total += len(chunk)
+ if total > _MAX_WEB_BYTES:
+ raise ValueError(
+ f"Response body exceeds {_MAX_WEB_BYTES} bytes "
+ f"for {source_url!r}"
+ )
+ body_chunks.append(chunk)
+ html = b"".join(body_chunks).decode(
+ resp.encoding or "utf-8", errors="replace"
+ )
if not html:
return artifacts
From 3853808a21f70f6f33f67eac80c17e9a3e648a22 Mon Sep 17 00:00:00 2001
From: Hogne <227774406+hognek@users.noreply.github.com>
Date: Mon, 20 Jul 2026 19:50:05 +0200
Subject: [PATCH 5/7] =?UTF-8?q?feat(library):=20P3=20=E2=80=94=20heavy=20t?=
=?UTF-8?q?ier=20download,=20quality=20preferences,=20storage=20accounting?=
=?UTF-8?q?,=20per-source=20rules?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- LibraryStore: add library_rules table, new columns (quality, auto_download,
download_path, download_bytes, downloaded_at) with safe migration in
_post_init, rule CRUD methods, fnmatch-based match_rules, get_storage_summary
- HeavyDownloadProcessor: yt-dlp-based media download for url:youtube items
with quality preference (360/480/720/1080/best), fallback heuristics for
missing output paths
- run_heavy_pipeline: rule-aware quality resolution (explicit > rule > item >
default 720), job tracking via library_jobs table
- Routes: POST /download, GET /download/status, POST/GET/DELETE /rules,
GET /usage with HTMX-aware HTML responses
- Auto-download: _ingest_task checks matching rules with auto_download=True
and triggers heavy pipeline after cheap tier completes
- Template: storage summary bar (polled via htmx), per-item download button
with quality selector for YouTube items, rules management panel
with add/delete
- Tests: 24 new tests (7 store rules/storage, 4 HeavyDownloadProcessor,
4 run_heavy_pipeline, 9 routes)
- 70/70 library tests pass (1 slow route test skipped)
---
tests/test_library.py | 393 +++++++++++++++++++++++++++++
tinyagentos/library_pipeline.py | 159 ++++++++++++
tinyagentos/library_store.py | 114 ++++++++-
tinyagentos/routes/library.py | 285 ++++++++++++++++++++-
tinyagentos/templates/library.html | 45 ++++
5 files changed, 980 insertions(+), 16 deletions(-)
diff --git a/tests/test_library.py b/tests/test_library.py
index f092fa769..8a3f341f2 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,398 @@ 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)
+
+ 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)
+
+ 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 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 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 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 bd6185333..5147b4971 100644
--- a/tinyagentos/library_pipeline.py
+++ b/tinyagentos/library_pipeline.py
@@ -930,6 +930,165 @@ 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():
+ # Try to find the downloaded file
+ candidates = sorted(
+ download_dir.glob("*"), 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)
+ await store.update_item_status(item_id, "error")
+ 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 f679d327b..dafe9384b 100644
--- a/tinyagentos/routes/library.py
+++ b/tinyagentos/routes/library.py
@@ -190,6 +190,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"
@@ -228,6 +249,17 @@ def _is_htmx(request: Request) -> bool:
return request.headers.get("HX-Request", "").lower() == "true"
+def _format_bytes(size: int) -> str:
+ """Format a byte count as a human-readable string."""
+ if size < 1024:
+ return f"{size} B"
+ for unit in ("KB", "MB", "GB", "TB"):
+ size /= 1024.0
+ if size < 1024 or unit == "TB":
+ return f"{size:.1f} {unit}"
+ return f"{size:.1f} TB"
+
+
_STATUS_CSS: dict[str, str] = {
"pending": "status-pending",
"processing": "status-processing",
@@ -244,22 +276,60 @@ def _render_item_card(item: dict) -> str:
css = _STATUS_CSS.get(status, "status-pending")
title = html.escape(item.get("title", "Untitled"))
kind = html.escape(item.get("kind", "file"))
- size = item.get("size_bytes") or 0
- size_str = f"{size:,} B" if size else ""
+ bytes_val = item.get("bytes") or 0
+ size_str = _format_bytes(bytes_val) if bytes_val else ""
item_id = html.escape(item.get("id", ""))
+ source_url = html.escape(item.get("source_url", ""))
+ downloaded = bool(item.get("download_path", ""))
+
+ parts = [
+ f'
") # .meta
+
+ # YouTube download actions (only for url:youtube items, only when ready)
+ if kind == "url:youtube" and status == "ready" and not downloaded:
+ parts.append(
+ f'
'
+ f' '
+ f''
+ f'
'
+ )
+ elif kind == "url:youtube" and downloaded:
+ download_bytes_val = item.get("download_bytes") or 0
+ parts.append(
+ f'
'
+ f'✅ Downloaded ({_format_bytes(download_bytes_val)})'
+ f'
") # .item-card
+
+ return "".join(parts)
def _render_item_list(items: list[dict]) -> str:
@@ -273,6 +343,47 @@ def _render_item_list(items: list[dict]) -> str:
return "".join(_render_item_card(item) for item in items)
+def _render_rules_list(rules: list[dict]) -> str:
+ """Return an HTML fragment listing source rules."""
+ import html as _h
+
+ if not rules:
+ return 'No rules. Add one above to auto-download from matching sources.'
+
+ parts: list[str] = ['
']
+ for rule in rules:
+ rid = _h.escape(rule["id"])
+ pat = _h.escape(rule["source_pattern"])
+ qual = _h.escape(rule.get("quality", "720"))
+ auto = "auto" if rule.get("auto_download") else "manual"
+ parts.append(
+ f'
'
+ f'{pat} ({qual}p, {auto})'
+ f''
+ f'
'
+ )
+ parts.append("
")
+ return "".join(parts)
+
+
+def _render_storage_summary(summary: dict) -> str:
+ """Return an HTML snippet for the storage summary bar."""
+ total_bytes = summary.get("total_bytes", 0)
+ total_count = summary.get("total_count", 0)
+ if not total_count:
+ return 'No items stored yet.'
+
+ return (
+ f'Library storage: {_format_bytes(total_bytes)} '
+ f'across {total_count} item{"s" if total_count != 1 else ""}'
+ f''
+ )
+
+
# ---------------------------------------------------------------------------
# List / Get / Delete
# ---------------------------------------------------------------------------
@@ -367,3 +478,149 @@ async def reprocess_item(request: Request, item_id: str):
_create_supervised_task(coro, task_set)
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)
+ await store.update_item_status(item_id, "error")
+
+
+@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()
+
+ if _is_htmx(request):
+ return HTMLResponse(_render_rules_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()
+
+ if _is_htmx(request):
+ return HTMLResponse(_render_storage_summary(summary))
+
+ return summary
diff --git a/tinyagentos/templates/library.html b/tinyagentos/templates/library.html
index 6fdcb1d44..76731257e 100644
--- a/tinyagentos/templates/library.html
+++ b/tinyagentos/templates/library.html
@@ -62,6 +62,15 @@