Exclude malformed URLs from catalog build instead of crashing

code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_catalog.py

@@ -94,6 +94,23 @@ def test_build_catalog_joins_and_canonicalizes(tmp_path):
     assert set(by_bot) == {"alpha", "beta"}
 
 
+def test_build_catalog_excludes_parse_raising_urls(tmp_path):
+    store, config = _seed(tmp_path)
+    # A bare "http://[" (junk extracted from a bot comment) makes urlsplit raise
+    # ValueError ("Invalid IPv6 URL"); it must count as malformed, not crash.
+    records = [
+        CitationRecord(url="http://[", run_id="r2", bot="alpha", question_id="100"),
+    ]
+    manifest_io.write_blob(store, "r2", records, config)
+
+    data = build_catalog(store, config)
+    assert data.excluded.get("malformed") == 1
+    assert "http://[" not in {s.canonical_url for s in data.sources}
+
+    summary = write_catalog(store, config)
+    assert summary.excluded.get("malformed") == 1
+
+
 def test_write_catalog_emits_views(tmp_path):
     store, config = _seed(tmp_path)
     summary = write_catalog(store, config)

forecasting_tools/agents_and_tools/source_archive/catalog.py

@@ -97,12 +97,19 @@ def _is_tool_call_only(source: "Source") -> bool:
 
 
 def is_search_url(url: str) -> bool:
-    host = urlsplit(url).netloc.lower()
+    try:
+        host = urlsplit(url).netloc.lower()
+    except ValueError:  # unparsable (see is_malformed_url) — not a search page
+        return False
     host = host[4:] if host.startswith("www.") else host
     return host in _SEARCH_HOSTS or host == "google.com" or host.startswith("google.")
 
 
 def is_malformed_url(url: str) -> bool:
+    try:
+        urlsplit(url)
+    except ValueError:  # e.g. a bare "http://[" — urlsplit: "Invalid IPv6 URL"
+        return True
     low = url.lower()
     return url.count("://") > 1 or any(m in low for m in _MALFORMED_MARKERS)
 
@@ -191,7 +198,10 @@ def question_url(self, qid: str) -> str | None:
 # Build (join manifests + index)
 # --------------------------------------------------------------------------- #
 def _domain(url: str) -> str:
-    host = urlsplit(url).netloc.lower()
+    try:
+        host = urlsplit(url).netloc.lower()
+    except ValueError:  # malformed URL — caller falls back to "(unknown)"
+        return ""
     return host[4:] if host.startswith("www.") else host