Q2-4: ingest and library nits (streaming, title unescape, JPEG modes, raise from, payload guard)

RED-FIRST: three acceptance tests were written and shown failing on
origin/dev before any production code was changed.

RED run (4 tests, 0 fixes applied):
```
FAILED tests/test_knowledge_ingest.py::test_download_article_unescapes_title_entities
FAILED tests/test_library.py::TestImageProcessor::test_process_pa_mode_image
FAILED tests/test_library.py::TestImageProcessor::test_process_la_mode_image
FAILED tests/test_agent_registry_store.py::TestTokenMinting::test_verify_non_dict_payload_raises
4 failed in 1.13s
```

GREEN run (after fixes):
```
4 passed in 0.84s
```

Changes (audit pass-2 card Q2-4, section 4.4):
- library_pipeline.py TextProcessor: stream file read/write in a single
  pass instead of loading the entire file into a str (avoids the 100 MB str
  copy); char_count, line_count, preview, and auto-title are computed
  incrementally while streaming.
- library_pipeline.py ImageProcessor: JPEG thumbnail conversion now handles
  LA, PA, I;16 and other non-RGB/L/CMYK Pillow modes instead of only RGBA/P.
- knowledge_ingest.py _download_article: unescape HTML entities in the
  title extracted from the <title> tag, matching library_pipeline.WebProcessor.
- x.py XWatchStore.create_watch: chain sqlite3.IntegrityError via
  raise from e.
- agent_registry_store.py verify_registry_token: isinstance payload dict
  guard with a clear ValueError when the JWT payload is not a JSON object.

Already fixed by prior commit (verified, no change needed):
- youtube.py / x.py: --dump-single-json (was --dump-json)
- youtube.py / x.py: asyncio.wait_for timeout on subprocess communicate()
- x.py fetch_tweet_ytdlp: raise from exc

Docs-Reviewed: no routes/, desktop app, or catalog manifest changes; these are internal library pipeline fixes with no agent-facing API surface

Author: jaylfc

changelog.d/tsk-55hgbu-library-nits.md

@@ -0,0 +1,7 @@
+### Fixed
+
+- TextProcessor now streams file reads instead of loading the entire file into a single `str`, eliminating the 100 MB `str` copy on large text files.
+- Article title extraction in `knowledge_ingest._download_article` now unescapes HTML entities (e.g. `&` becomes `&`), matching the behaviour already present in `library_pipeline.WebProcessor`.
+- ImageProcessor JPEG thumbnail conversion now handles `LA`, `PA`, `I;16` and other non-RGB/L/CMYK Pillow modes instead of raising on them.
+- `x.py` `create_watch` now chains the `sqlite3.IntegrityError` via `raise ... from e`.
+- `verify_registry_token` now raises `ValueError` with a clear message when the JWT payload is not a JSON object (dict).

tests/test_agent_registry_store.py

@@ -278,6 +278,27 @@ def test_verify_truncated_token_raises(self, signing_keypair):
         with pytest.raises(ValueError):
             verify_registry_token("one", pub)
 
+    def test_verify_non_dict_payload_raises(self, signing_keypair):
+        """RED-FIRST: a token whose payload JSON decodes to a non-dict
+        (e.g. a JSON list) must raise ValueError mentioning the field name,
+        not silently return the non-dict value."""
+        from cryptography.hazmat.primitives.serialization import load_pem_private_key
+
+        priv, pub = signing_keypair
+        private_key = load_pem_private_key(priv, password=None)
+
+        header = _b64url_encode(
+            json.dumps({"alg": "EdDSA", "typ": "JWT"}, separators=(",", ":")).encode()
+        )
+        # Payload is a JSON array, NOT a dict
+        payload = _b64url_encode(json.dumps([1, 2, 3]).encode())
+        signing_input = f"{header}.{payload}".encode()
+        signature = _b64url_encode(private_key.sign(signing_input))
+        token = f"{header}.{payload}.{signature}"
+
+        with pytest.raises(ValueError, match="payload"):
+            verify_registry_token(token, pub)
+
     def test_token_has_jti(self, signing_keypair):
         priv, pub = signing_keypair
         token = mint_registry_token("agent-006", priv)

tests/test_knowledge_ingest.py

@@ -715,3 +715,46 @@ async def test_embed_failure_sets_partial_status(store):
     assert item["status"] == "partial", (
         f"Expected 'partial' status on embed failure, got '{item['status']}'"
     )
+
+
+# ------------------------------------------------------------------
+# Q2-4 RED-FIRST: title HTML entities must be unescaped
+# ------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_download_article_unescapes_title_entities(store):
+    """RED-FIRST: a <title> containing HTML entities (e.g. &amp;) must be
+    unescaped to the bare character in the returned title.
+    &amp; must become &."""
+    html = (
+        "<html><head><title>Test &amp; Demo Title</title></head>"
+        "<body><article>"
+        "<p>This is a long enough article body for readability extraction.</p>"
+        "</article></body></html>"
+    )
+    resp = _TrackedResponse([html.encode("utf-8")])
+
+    mock_http = AsyncMock()
+    mock_http.get = AsyncMock(return_value=resp)
+
+    notif = AsyncMock()
+    notif.emit_event = AsyncMock()
+    cat_engine = AsyncMock()
+    cat_engine.categorise = AsyncMock(return_value=[])
+
+    pipeline = IngestPipeline(
+        store=store,
+        http_client=mock_http,
+        fetch_client=mock_http,
+        notifications=notif,
+        category_engine=cat_engine,
+    )
+
+    content, title, author, metadata = await pipeline._download_article(
+        "https://example.com/test", "", {}
+    )
+
+    assert title == "Test & Demo Title", (
+        f"Expected unescaped title 'Test & Demo Title', got: {title!r}"
+    )

tests/test_library.py

@@ -356,6 +356,54 @@ async def test_process_missing_image(self, lib_store, storage_dir):
         artifacts = await proc.process(item)
         assert len(artifacts) == 0
 
+    @pytest.mark.asyncio
+    async def test_process_pa_mode_image(self, lib_store, storage_dir):
+        """RED-FIRST: a PA-mode (palette + alpha) image must convert to JPEG
+        without raising, producing a thumbnail artifact."""
+        from PIL import Image
+
+        file_path = storage_dir / "test_pa.tiff"
+        img = Image.new("PA", (100, 50), color=(128, 255))
+        img.save(file_path, format="TIFF")
+
+        item_id = await lib_store.create_item(
+            kind="image", title="test_pa.tiff", storage_path=str(file_path)
+        )
+        item = await lib_store.get_item(item_id)
+
+        proc = ImageProcessor(lib_store, storage_dir)
+        artifacts = await proc.process(item)
+
+        kinds = {a["kind"] for a in artifacts}
+        assert "thumbnail" in kinds, (
+            f"PA-mode image should produce a thumbnail, got kinds: {kinds}"
+        )
+        thumb_art = [a for a in artifacts if a["kind"] == "thumbnail"][0]
+        assert Path(thumb_art["path"]).exists()
+
+    @pytest.mark.asyncio
+    async def test_process_la_mode_image(self, lib_store, storage_dir):
+        """RED-FIRST: an LA-mode (grayscale + alpha) PNG must convert to JPEG
+        without raising, producing a thumbnail artifact."""
+        from PIL import Image
+
+        file_path = storage_dir / "test_la.png"
+        img = Image.new("LA", (100, 50), color=(128, 255))
+        img.save(file_path, format="PNG")
+
+        item_id = await lib_store.create_item(
+            kind="image", title="test_la.png", storage_path=str(file_path)
+        )
+        item = await lib_store.get_item(item_id)
+
+        proc = ImageProcessor(lib_store, storage_dir)
+        artifacts = await proc.process(item)
+
+        kinds = {a["kind"] for a in artifacts}
+        assert "thumbnail" in kinds, (
+            f"LA-mode image should produce a thumbnail, got kinds: {kinds}"
+        )
+
 
 # ---------------------------------------------------------------------------
 # run_pipeline

tinyagentos/agent_registry_store.py

@@ -394,6 +394,10 @@ def verify_registry_token(token: str, public_key_pem: bytes) -> dict:
         raise ValueError("token signature verification failed") from None
 
     payload = json.loads(_b64url_decode(payload_b64))
+    if not isinstance(payload, dict):
+        raise ValueError(
+            f"payload must be a JSON object, got {type(payload).__name__}"
+        )
     return payload
 
 

tinyagentos/knowledge_fetchers/x.py

@@ -299,8 +299,8 @@ def create_watch(
                 (handle.lstrip("@"), filters_json, frequency, created_at),
             )
             conn.commit()
-        except sqlite3.IntegrityError:
-            raise ValueError(f"Watch for @{handle} already exists")
+        except sqlite3.IntegrityError as e:
+            raise ValueError(f"Watch for @{handle} already exists") from e
         return self.get_watch(handle.lstrip("@"))  # type: ignore[return-value]
 
     def list_watches(self) -> list[dict]:

tinyagentos/knowledge_ingest.py

@@ -431,7 +431,7 @@ async def _download_article(
         if not title:
             m = re.search(r"<title[^>]*>([^<]+)</title>", html, re.IGNORECASE)
             if m:
-                title = m.group(1).strip()
+                title = _html_mod.unescape(m.group(1)).strip()
         return content, title, "", metadata
 
     # ------------------------------------------------------------------

tinyagentos/library_pipeline.py

@@ -154,21 +154,32 @@ async def process(self, item: dict) -> list[dict]:
             return artifacts
 
         try:
-            text = p.read_text(encoding="utf-8", errors="replace")
+            char_count = 0
+            line_count = 1
+            preview = ""
+            text_dir = self.storage_dir / "text"
+            text_dir.mkdir(parents=True, exist_ok=True)
+            text_path = text_dir / f"{item_id}.txt"
+            with open(p, "r", encoding="utf-8", errors="replace") as src:
+                with open(text_path, "w", encoding="utf-8") as dst:
+                    while True:
+                        chunk = src.read(8192)
+                        if not chunk:
+                            break
+                        dst.write(chunk)
+                        char_count += len(chunk)
+                        line_count += chunk.count("\n")
+                        if len(preview) < 200:
+                            preview += chunk
+                            preview = preview[:200]
         except Exception:
             logger.warning("Text processor: could not read %s", storage_path,
                            exc_info=True)
             return artifacts
 
-        # Write extracted text as an artifact
-        text_dir = self.storage_dir / "text"
-        text_dir.mkdir(parents=True, exist_ok=True)
-        text_path = text_dir / f"{item_id}.txt"
-        text_path.write_text(text, encoding="utf-8")
-
         text_meta = {
-            "char_count": len(text),
-            "line_count": text.count("\n") + 1,
+            "char_count": char_count,
+            "line_count": line_count,
             "source_url": item.get("source_url", ""),
             "processed_at": time.time(),
             "processor": "TextProcessor/v1",
@@ -179,14 +190,13 @@ async def process(self, item: dict) -> list[dict]:
         artifacts.append({"kind": "text", "path": str(text_path), "meta": text_meta})
 
         # Store a preview (first 200 chars)
-        preview = text[:200]
         meta = json.loads(item.get("meta_json", "{}"))
         meta["preview"] = preview
         await self.store.update_item(item_id, meta_json=meta)
 
         # Auto-title from content if no title
         if not item.get("title"):
-            title = text.strip().split("\n", 1)[0][:100]
+            title = preview.strip().split("\n", 1)[0][:100]
             if title:
                 await self.store.update_item(item_id, title=title)
 
@@ -295,8 +305,10 @@ async def process(self, item: dict) -> list[dict]:
                 thumb_path = thumb_dir / f"{item_id}_thumb.jpg"
 
                 img.thumbnail((320, 320))
-                # Convert to RGB if needed (e.g. RGBA/PNG → JPEG)
-                if img.mode in ("RGBA", "P"):
+                # Convert to RGB if needed (e.g. RGBA/PNG → JPEG).
+                # JPEG supports only "L", "RGB", "CMYK"; all other modes
+                # (including LA, PA, I;16) must be converted first.
+                if img.mode not in ("RGB", "L", "CMYK"):
                     img = img.convert("RGB")
                 img.save(thumb_path, "JPEG", quality=75)