diff --git a/tests/test_library.py b/tests/test_library.py index 37100dab6..c8eace7e7 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -5,6 +5,7 @@ import json import tempfile from pathlib import Path +from unittest.mock import patch, MagicMock, AsyncMock import pytest import pytest_asyncio @@ -14,6 +15,8 @@ ImageProcessor, PdfProcessor, TextProcessor, + WebProcessor, + YouTubeProcessor, detect_kind, run_pipeline, ) @@ -338,7 +341,7 @@ async def test_process_image(self, lib_store, storage_dir): assert "metadata" in kinds assert "thumbnail" in kinds - thumb_art = [a for a in artifacts if a["kind"] == "thumbnail"][0] + thumb_art = next(a for a in artifacts if a["kind"] == "thumbnail") assert Path(thumb_art["path"]).exists() @pytest.mark.asyncio @@ -353,6 +356,272 @@ async def test_process_missing_image(self, lib_store, storage_dir): assert len(artifacts) == 0 +class TestYouTubeProcessor: + @pytest.mark.asyncio + async def test_process_youtube_url(self, lib_store, storage_dir): + """YouTube processor extracts metadata, transcript, thumbnail, chapters.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://www.youtube.com/watch?v=test123", + title="", + ) + item = await lib_store.get_item(item_id) + + proc = YouTubeProcessor(lib_store, storage_dir) + + mock_result = { + "title": "Test Video", + "author": "TestChannel", + "content": "This is the transcript text with enough content to verify.", + "thumbnail": None, + "metadata": { + "video_id": "test123", + "channel": "TestChannel", + "views": 1000, + "likes": 50, + "duration": 120.5, + "upload_date": "20250101", + "chapters": [ + {"title": "Intro", "start_time": 0.0, "end_time": 30.0}, + {"title": "Main", "start_time": 30.0, "end_time": 90.0}, + ], + }, + } + + with ( + patch( + "tinyagentos.knowledge_fetchers.youtube.fetch", + _async_return(mock_result), + ), + patch( + "tinyagentos.knowledge_fetchers.youtube.format_timestamp", + side_effect=lambda s: f"{int(s // 60):02d}:{int(s % 60):02d}", + ), + ): + artifacts = await proc.process(item) + + kinds = {a["kind"] for a in artifacts} + assert "metadata" in kinds + assert "transcript" in kinds + assert "chapters" in kinds + + updated = await lib_store.get_item(item_id) + meta = json.loads(updated["meta_json"]) + assert meta["video_id"] == "test123" + assert meta["channel"] == "TestChannel" + assert meta["duration"] == 120.5 + assert "preview" in meta + + updated_title = updated.get("title", "") + assert updated_title == "Test Video", ( + f"YouTube processor should set title from video metadata; " + f"got {updated_title!r}" + ) + + @pytest.mark.asyncio + async def test_process_youtube_url_no_source(self, lib_store, storage_dir): + """YouTube processor returns empty when source_url is missing.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="", + title="", + ) + item = await lib_store.get_item(item_id) + + proc = YouTubeProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert artifacts == [] + + @pytest.mark.asyncio + async def test_youtube_handles_missing_thumbnail(self, lib_store, storage_dir): + """YouTube processor does not error when thumbnail path is absent.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=no-thumb", + ) + item = await lib_store.get_item(item_id) + + proc = YouTubeProcessor(lib_store, storage_dir) + + mock_result = { + "title": "No Thumbnail Video", + "author": "TestChannel", + "content": "Some transcript text.", + "thumbnail": "/nonexistent/path/thumb.png", + "metadata": {"video_id": "no-thumb", "channel": "TestChannel"}, + } + + with patch( + "tinyagentos.knowledge_fetchers.youtube.fetch", + _async_return(mock_result), + ): + artifacts = await proc.process(item) + + # Should not have a thumbnail artifact + thumb_artifacts = [a for a in artifacts if a["kind"] == "thumbnail"] + assert len(thumb_artifacts) == 0 + + @pytest.mark.asyncio + async def test_youtube_pipeline_integration(self, lib_store, storage_dir): + """run_pipeline routes url:youtube items to YouTubeProcessor.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=pipeline-test", + ) + + mock_result = { + "title": "Pipeline Video", + "author": "PipelineChannel", + "content": "Pipeline transcript.", + "thumbnail": None, + "metadata": { + "video_id": "pipeline-test", + "channel": "PipelineChannel", + }, + } + + with patch( + "tinyagentos.knowledge_fetchers.youtube.fetch", + _async_return(mock_result), + ): + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + artifacts = await lib_store.get_artifacts(item_id) + artifact_kinds = {a["kind"] for a in artifacts} + # FileProcessor runs first (no storage_path → empty), then YouTubeProcessor + assert "metadata" in artifact_kinds + assert "transcript" in artifact_kinds + + +class TestWebProcessor: + @pytest.mark.asyncio + async def test_process_web_url(self, lib_store, storage_dir): + """Web processor fetches HTML, extracts text, stores artifacts.""" + html = ( + "Test Page" + "

This is the main article text " + "with enough content to pass the readability minimum " + "character count for extraction purposes now.

" + "

Second paragraph with more detailed content about " + "the topic being discussed on this test page.

" + "" + ) + + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/article", + title="", + ) + item = await lib_store.get_item(item_id) + + proc = WebProcessor(lib_store, storage_dir) + + mock_resp = _mock_httpx_response(html, 200) + with ( + patch("httpx.AsyncClient") as mock_client_cls, + patch( + "tinyagentos.routes.desktop_browser.ssrf.validate_url_or_raise", + ), + ): + _mock_httpx_stream(mock_client_cls, mock_resp) + artifacts = await proc.process(item) + + kinds = {a["kind"] for a in artifacts} + assert "metadata" in kinds + assert "text" in kinds + + text_artifacts = [a for a in artifacts if a["kind"] == "text"] + assert len(text_artifacts) == 1 + text_path = Path(text_artifacts[0]["path"]) + assert text_path.exists() + + updated = await lib_store.get_item(item_id) + meta = json.loads(updated["meta_json"]) + assert "preview" in meta + + @pytest.mark.asyncio + async def test_process_web_url_no_source(self, lib_store, storage_dir): + """Web processor returns empty when source_url is missing.""" + item_id = await lib_store.create_item( + kind="url:web", source_url="", title="", + ) + item = await lib_store.get_item(item_id) + + proc = WebProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert artifacts == [] + + @pytest.mark.asyncio + async def test_web_extracts_title_from_html(self, lib_store, storage_dir): + """Web processor auto-titles from tag when item has no title.""" + html = ( + "<html><head><title>Auto Title Here" + "

This article has enough text content " + "to ensure that the readability extractor returns a full " + "result instead of falling back to the simple tag stripper " + "which would otherwise happen for very short pages.

" + "" + ) + + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/titled", + title="https://example.com/titled", + ) + item = await lib_store.get_item(item_id) + + proc = WebProcessor(lib_store, storage_dir) + + mock_resp = _mock_httpx_response(html, 200) + with ( + patch("httpx.AsyncClient") as mock_client_cls, + patch( + "tinyagentos.routes.desktop_browser.ssrf.validate_url_or_raise", + ), + ): + _mock_httpx_stream(mock_client_cls, mock_resp) + await proc.process(item) + + updated = await lib_store.get_item(item_id) + assert "Auto Title Here" in updated["title"] + + @pytest.mark.asyncio + async def test_web_pipeline_integration(self, lib_store, storage_dir): + """run_pipeline routes url:web items to WebProcessor.""" + html = ( + "Pipeline Web" + "

Pipeline test content that is sufficiently " + "long to pass the readability minimum character count." + "

" + ) + + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/pipeline-web", + ) + + mock_resp = _mock_httpx_response(html, 200) + with ( + patch("httpx.AsyncClient") as mock_client_cls, + patch( + "tinyagentos.routes.desktop_browser.ssrf.validate_url_or_raise", + ), + ): + _mock_httpx_stream(mock_client_cls, mock_resp) + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + artifacts = await lib_store.get_artifacts(item_id) + artifact_kinds = {a["kind"] for a in artifacts} + assert "metadata" in artifact_kinds + assert "text" in artifact_kinds + + # --------------------------------------------------------------------------- # run_pipeline # --------------------------------------------------------------------------- @@ -437,7 +706,6 @@ async def test_handoff_files_copied(self, lib_store, storage_dir): @pytest.mark.asyncio async def test_handoff_with_qmd(self, lib_store, storage_dir): """Handoff indexes into taosmd when the API is reachable.""" - from unittest.mock import AsyncMock, patch file_path = storage_dir / "notes.txt" file_path.write_text("hello from library") @@ -455,7 +723,6 @@ async def test_handoff_with_qmd(self, lib_store, storage_dir): taosmd_token = "test-admin-token" # Mock httpx.AsyncClient to simulate a working taosmd - from unittest.mock import AsyncMock, MagicMock mock_client = MagicMock() mock_client.__aenter__ = AsyncMock(return_value=mock_client) @@ -613,12 +880,12 @@ async def test_filter_by_kind(self, client): # -- Auth: all endpoints require a session (CSRF is bypassed in tests, but # unauthenticated requests still hit the global auth middleware). Test - # that the 6 new endpoints return 401 when no session cookie/header is + # that the 5 library endpoints return 401 when no session cookie/header is # present. @pytest.mark.asyncio async def test_unauth_library_endpoints(self, app): - """All 6 library endpoints return 401 without authentication.""" + """All 5 library endpoints return 401 without authentication.""" from httpx import ASGITransport, AsyncClient async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as unauth_client: @@ -655,9 +922,16 @@ async def test_reprocess_idempotent(self, client, tmp_path): assert resp.status_code == 202 item_id = resp.json()["item_id"] - # Wait for pipeline to finish (background task) + # Wait for pipeline to finish (background task) — poll status, not a fixed sleep import asyncio - await asyncio.sleep(0.5) + for _ in range(50): # up to 5 s + resp = await client.get(f"/api/library/items/{item_id}") + status = resp.json()["item"].get("status") + if status in ("ready", "error"): + break + await asyncio.sleep(0.1) + else: + pytest.fail("Pipeline did not finish within 5 s") # Check initial artifact count resp = await client.get(f"/api/library/items/{item_id}") @@ -668,7 +942,15 @@ async def test_reprocess_idempotent(self, client, tmp_path): # First reprocess resp = await client.post(f"/api/library/items/{item_id}/reprocess") assert resp.status_code == 202 - await asyncio.sleep(0.5) + # Poll until reprocess finishes + for _ in range(50): # up to 5 s + resp = await client.get(f"/api/library/items/{item_id}") + status = resp.json()["item"].get("status") + if status in ("ready", "error"): + break + await asyncio.sleep(0.1) + else: + pytest.fail("Reprocess did not finish within 5 s") # After first reprocess — exact same artifact count, not doubled. resp = await client.get(f"/api/library/items/{item_id}") @@ -698,7 +980,15 @@ async def test_reprocess_while_processing_returns_409(self, client, app): ) assert resp.status_code == 202 item_id = resp.json()["item_id"] - await asyncio.sleep(0.5) + # Poll until pipeline finishes — not a fixed sleep + for _ in range(50): # up to 5 s + resp = await client.get(f"/api/library/items/{item_id}") + status = resp.json()["item"].get("status") + if status in ("ready", "error"): + break + await asyncio.sleep(0.1) + else: + pytest.fail("Pipeline did not finish within 5 s") # Force status to "processing" to simulate an in-flight pipeline store = app.state.library_store @@ -733,3 +1023,46 @@ def _create_test_image(path: Path): from PIL import Image img = Image.new("RGB", (100, 50), color="blue") img.save(path) + + +def _async_return(value): + """Return an async callable that returns ``value`` (for mocking async functions).""" + async def _inner(*args, **kwargs): + return value + return _inner + + +def _make_stream_ctx(resp): + """Return a mock async context manager whose __aenter__ yields *resp*.""" + ctx = MagicMock() + ctx.__aenter__ = _async_return(resp) + async def _aexit(*args, **kwargs): + return None + ctx.__aexit__ = _aexit + return ctx + + +def _mock_httpx_stream(mock_client_cls, resp): + """Configure the mocked httpx.AsyncClient to use client.stream(). + + After this call, ``client.stream(\"GET\", url)`` returns an async context + manager whose ``__aenter__`` yields *resp*. + """ + client = mock_client_cls.return_value.__aenter__.return_value + client.stream = MagicMock(return_value=_make_stream_ctx(resp)) + + +def _mock_httpx_response(html: str, status_code: int = 200): + """Return a mock httpx Response with the given HTML body.""" + mock = MagicMock() + mock.text = html + mock.status_code = status_code + mock.is_redirect = False + mock.encoding = "utf-8" + mock.headers = {"content-type": "text/html; charset=utf-8"} + + async def _aiter_bytes(_chunk_size: int = 8192): + yield html.encode("utf-8") + + mock.aiter_bytes = _aiter_bytes + return mock diff --git a/tinyagentos/library_collections.py b/tinyagentos/library_collections.py index c4cddb845..c224432b9 100644 --- a/tinyagentos/library_collections.py +++ b/tinyagentos/library_collections.py @@ -325,11 +325,12 @@ async def _set_retryable() -> None: # 4. Link the collection to the library item. try: - await http_client.post( + link_resp = await http_client.post( f"{base_url}/collections/{collection_id}/link", json={"type": "taos", "id": item_id}, headers=auth_headers, ) + link_resp.raise_for_status() logger.debug( "Linked collection %s to library item %s", collection_id, item_id, diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py index e2ef7e185..74e2f5ba3 100644 --- a/tinyagentos/library_pipeline.py +++ b/tinyagentos/library_pipeline.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import json import logging import mimetypes @@ -257,6 +258,289 @@ async def process(self, item: dict) -> list[dict]: return artifacts +class YouTubeProcessor(Processor): + """YouTube URL processor — cheap tier: metadata, thumbnail, transcript, chapters. + + Uses yt-dlp via the knowledge_fetchers.youtube module to fetch video + metadata and captions without downloading the video file. Produces + artifacts that flow into taosmd collections for agent querying. + + The cheap tier (per the design doc, docs/design/library-app.md section 4) + covers steps 1-4: canonical link, title, channel, description, thumbnail, + duration, upload date, subtitles/transcript, chapters. + """ + + 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 + + # Only catch ImportError (missing yt-dlp). Let fetch errors + # propagate so run_pipeline marks the item as "error" — a failed + # yt-dlp invocation must not silently look successful. + from tinyagentos.knowledge_fetchers.youtube import ( + fetch, + format_timestamp, + ) + + media_dir = self.storage_dir / "youtube" + result = await asyncio.wait_for( + fetch(source_url, media_dir=media_dir), + timeout=120, # yt-dlp can hang on network stalls / large channels + ) + + title = result.get("title", "") + if title and not item.get("title"): + await self.store.update_item(item_id, title=title) + + meta = result.get("metadata", {}) + + # Update item meta_json with structured video metadata + stored_meta = json.loads(item.get("meta_json", "{}")) + stored_meta.update({ + "video_id": meta.get("video_id", ""), + "channel": meta.get("channel", ""), + "duration": meta.get("duration"), + "views": meta.get("views"), + "upload_date": meta.get("upload_date", ""), + }) + await self.store.update_item(item_id, meta_json=stored_meta) + + # Artifact: metadata + await self.store.add_artifact( + item_id, kind="metadata", path=source_url, meta=meta, + ) + artifacts.append({"kind": "metadata", "path": source_url, "meta": meta}) + + # Artifact: thumbnail (if downloaded) + thumbnail = result.get("thumbnail") + if thumbnail and Path(thumbnail).exists(): + await self.store.add_artifact( + item_id, kind="thumbnail", path=thumbnail, + meta={"source": "youtube"}, + ) + artifacts.append({ + "kind": "thumbnail", "path": thumbnail, + "meta": {"source": "youtube"}, + }) + + # Artifact: transcript + content = result.get("content", "") + if content: + transcript_dir = self.storage_dir / "transcripts" + transcript_dir.mkdir(parents=True, exist_ok=True) + transcript_path = transcript_dir / f"{item_id}_transcript.txt" + transcript_path.write_text(content, encoding="utf-8") + + transcript_meta = { + "char_count": len(content), + "language": "en", + } + await self.store.add_artifact( + item_id, kind="transcript", path=str(transcript_path), + meta=transcript_meta, + ) + artifacts.append({ + "kind": "transcript", "path": str(transcript_path), + "meta": transcript_meta, + }) + + # Store preview for item card + preview = content[:200] + stored_meta["preview"] = preview + await self.store.update_item(item_id, meta_json=stored_meta) + + # Artifact: chapters (if available) + chapters = meta.get("chapters", []) + if chapters: + chapters_lines: list[str] = [] + for ch in chapters: + ts = format_timestamp(ch.get("start_time", 0)) + ch_title = ch.get("title", "") + chapters_lines.append(f"[{ts}] {ch_title}") + + chapters_text = "\n".join(chapters_lines) + chapters_dir = self.storage_dir / "chapters" + chapters_dir.mkdir(parents=True, exist_ok=True) + chapters_path = chapters_dir / f"{item_id}_chapters.txt" + chapters_path.write_text(chapters_text, encoding="utf-8") + + await self.store.add_artifact( + item_id, kind="chapters", path=str(chapters_path), + meta={"count": len(chapters)}, + ) + artifacts.append({ + "kind": "chapters", "path": str(chapters_path), + "meta": {"count": len(chapters)}, + }) + + return artifacts + + +class WebProcessor(Processor): + """Generic web-page URL processor — extracts readable text from HTML. + + Fetches the URL (SSRF-guarded against loopback/link-local/private hosts), + then extracts the main content using readability-lxml. Falls back to a + simple tag-stripping approach when readability-lxml is not installed. + + Produces a text artifact suitable for taosmd collection indexing so that + agents granted the collection can query the page content. + """ + + 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 + + # 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, + ) + + _MAX_WEB_REDIRECTS = 5 + _MAX_WEB_BYTES = 10 * 1024 * 1024 # 10 MB + + current_url = source_url + async with httpx.AsyncClient( + timeout=httpx.Timeout(30), + follow_redirects=False, + ) as client: + html = "" + for _hop in range(_MAX_WEB_REDIRECTS + 1): + validate_url_or_raise(current_url) + async with client.stream("GET", current_url) as resp: + if resp.is_redirect and resp.headers.get("location"): + current_url = urljoin( + current_url, resp.headers["location"], + ) + continue + + resp.raise_for_status() + + # Read body with a size cap to avoid OOM. + 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} " + f"bytes for {source_url!r}" + ) + body_chunks.append(chunk) + html = b"".join(body_chunks).decode( + resp.encoding or "utf-8", errors="replace", + ) + break + else: + raise SsrfBlockedError( + f"too many redirects fetching {source_url!r}" + ) + + if not html: + return artifacts + + # Extract readable text + content = _extract_readable_text(html, source_url) + + # Extract title from tag if item has no title + import html as _html_mod + title = item.get("title", "") + if not title or title == source_url: + import re + m = re.search( + r"<title[^>]*>([^<]+)", html, re.IGNORECASE, + ) + if m: + title = _html_mod.unescape(m.group(1)).strip()[:200] + await self.store.update_item(item_id, title=title) + + # Artifact: metadata + page_meta = { + "char_count": len(content), + "content_type": resp.headers.get("content-type", ""), + "status_code": resp.status_code, + } + await self.store.add_artifact( + item_id, kind="metadata", path=source_url, meta=page_meta, + ) + artifacts.append({ + "kind": "metadata", "path": source_url, "meta": page_meta, + }) + + # Artifact: extracted text + if content: + text_dir = self.storage_dir / "text" + text_dir.mkdir(parents=True, exist_ok=True) + text_path = text_dir / f"{item_id}_web.txt" + text_path.write_text(content, encoding="utf-8") + + text_meta = { + "char_count": len(content), + "source": "readability", + } + await self.store.add_artifact( + item_id, kind="text", path=str(text_path), meta=text_meta, + ) + artifacts.append({ + "kind": "text", "path": str(text_path), "meta": text_meta, + }) + + # Store preview + preview = content[:200] + stored_meta = json.loads(item.get("meta_json", "{}")) + stored_meta["preview"] = preview + await self.store.update_item(item_id, meta_json=stored_meta) + + return artifacts + + +def _extract_readable_text(html: str, source_url: str = "") -> str: + """Extract the main readable content from an HTML page. + + Uses readability-lxml when available; falls back to simple tag-stripping. + """ + try: + from readability import Document + doc = Document(html) + content = doc.summary() + # Strip remaining HTML from readability output + import re + text = re.sub(r"<[^>]+>", " ", content) + text = re.sub(r"\s+", " ", text).strip() + if len(text) >= 100: + return text + except ImportError: + logger.debug("readability-lxml not installed — using simple extractor") + except Exception: + logger.warning("readability extraction failed for %s", source_url, + exc_info=True) + + # Fallback: simple tag-stripping (from knowledge_ingest._extract_text_readability) + import re + cleaned = re.sub( + r"<(script|style)[^>]*>.*?", "", html, + flags=re.DOTALL | re.IGNORECASE, + ) + text = re.sub(r"<[^>]+>", " ", cleaned) + text = re.sub(r"\s+", " ", text).strip() + return text + + class ImageProcessor(Processor): """Image processor — records dimensions, creates thumbnail. @@ -328,6 +612,8 @@ async def process(self, item: dict) -> list[dict]: "text": TextProcessor, "pdf": PdfProcessor, "image": ImageProcessor, + "url:youtube": YouTubeProcessor, + "url:web": WebProcessor, } @@ -363,23 +649,29 @@ async def run_pipeline( kind = item["kind"] - # URL-only items (no storage_path) are stored as references — the pipeline - # records a reference metadata artifact but does not fetch remote content - # (future: WebFetcherProcessor, #2078). The item gets a "reference" artifact so it - # is not silently empty. + # URL-only items (no storage_path) are stored as references only when + # no kind-specific processor is registered to fetch them. url:youtube + # and url:web have dedicated processors; other URL kinds fall through + # to the reference-only path (future: WebFetcherProcessor, #2078). if not item.get("storage_path") and item.get("source_url"): - logger.info( - "Library pipeline: URL-only item %s (%s) — stored as reference, not fetched", - item_id, kind, - ) - ref_meta = { - "source_url": item["source_url"], - "kind": kind, - "note": "Reference-only item — content not fetched (TODO: WebFetcherProcessor)", - } - await store.add_artifact( - item_id, kind="reference", path="", meta=ref_meta, - ) + proc_cls = _PROCESSORS.get(kind) + if proc_cls is None or proc_cls is FileProcessor: + logger.info( + "Library pipeline: URL-only item %s (%s) — no processor registered, " + "stored as reference", + item_id, kind, + ) + ref_meta = { + "source_url": item["source_url"], + "kind": kind, + "note": ( + "Reference-only item — no processor registered for " + f"kind={kind} (TODO: WebFetcherProcessor, #2078)" + ), + } + await store.add_artifact( + item_id, kind="reference", path="", meta=ref_meta, + ) # If item has a storage_path that points to a missing file, fail early # (dropped/moved/corrupt source must not silently look successful). diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py index 067a98ebc..420fff7dd 100644 --- a/tinyagentos/routes/library.py +++ b/tinyagentos/routes/library.py @@ -31,6 +31,10 @@ # garbage-collected when request.app.state._background_tasks is absent. _background_tasks: set[asyncio.Task] = set() +# Guard lazy LibraryStore init against concurrent requests. +# Same pattern as _config_lock in tinyagentos/config.py. +_store_init_lock = asyncio.Lock() + def _track_background_task(coro) -> asyncio.Task: """Create a task, store it in ``_background_tasks``, and auto-discard on done.""" @@ -65,13 +69,16 @@ async def _get_library_store(request: Request): """Get the LibraryStore from app.state (lazily initialised).""" store = getattr(request.app.state, "library_store", None) if store is None: - from tinyagentos.library_store import LibraryStore - - data_dir = getattr(request.app.state, "data_dir", None) - base = Path(data_dir) if data_dir else Path(__file__).parent.parent.parent / "data" - store = LibraryStore(base / "library.db") - await store.init() - request.app.state.library_store = store + async with _store_init_lock: + store = getattr(request.app.state, "library_store", None) + if store is None: + from tinyagentos.library_store import LibraryStore + + data_dir = getattr(request.app.state, "data_dir", None) + base = Path(data_dir) if data_dir else Path(__file__).parent.parent.parent / "data" + store = LibraryStore(base / "library.db") + await store.init() + request.app.state.library_store = store return store @@ -171,6 +178,14 @@ async def _ingest_task(app, item_id: str, store, storage_dir: Path) -> None: await store.update_item_status(item_id, "error") return + # Re-acquire processing state so reprocess sees us as busy during the + # collections handoff window. If we lose the CAS race, another pipeline + # already claimed the item — abort cleanly (it will do its own handoff). + if not await store.try_update_item_status( + item_id, "processing", if_not_in=("pending", "processing") + ): + return + # Collections handoff after successful pipeline try: collections_dir = storage_dir.parent / "collections" @@ -198,8 +213,10 @@ async def _ingest_task(app, item_id: str, store, storage_dir: Path) -> None: "(no text artifacts or taosmd unavailable)", item_id, ) + await store.update_item_status(item_id, "ready") except Exception: logger.exception("Collections handoff failed for item %s", item_id) + await store.update_item_status(item_id, "error") def _sanitise_filename(name: str) -> str: @@ -338,8 +355,8 @@ async def reprocess_item(request: Request, item_id: str): 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. + # re-queue the pipeline now. If scheduling fails, mark as error — artifacts + # were already deleted so rolling back to ready would leave a data-less item. try: task_set = getattr(request.app.state, "_background_tasks", None) coro = _ingest_task(request.app, item_id, store, storage_dir) @@ -349,7 +366,7 @@ async def reprocess_item(request: Request, item_id: str): _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") + await store.update_item_status(item_id, "error") return JSONResponse( {"error": "Failed to schedule reprocess"}, status_code=500, )