@@ -23,7 +23,8 @@
     BrowserConfig,
     MemoryAdaptiveDispatcher,
     RateLimiter, 
-    LLMConfig
+    LLMConfig,
+    CrawlResult,
 )
 from crawl4ai.async_configs import Provenance, UntrustedConfigError
 from hook_registry import build_declarative_hooks, HookValidationError
@@ -647,14 +648,26 @@ async def stream_results(crawler: AsyncWebCrawler, results_gen: AsyncGenerator)
             await _dispose_crawler(crawler)
 
 
-def _normalize_and_validate_seeds(urls: List[str]) -> List[str]:
+def _normalize_and_validate_seeds(urls: List[str]) -> Tuple[List[str], List[CrawlResult]]:
     """Prefix bare hosts with https:// and SSRF-validate every seed URL's
     destination. Shared by the streaming and non-streaming crawl handlers so a
-    new entry point cannot silently skip the destination check."""
+    new entry point cannot silently skip the destination check.
+
+    Returns (allowed, refused): a refused seed becomes a failed CrawlResult so
+    one bad seed does not sink the batch. If no seed is allowed, the first
+    refusal is raised (400), as before."""
     urls = [('https://' + url) if not url.startswith(('http://', 'https://')) and not url.startswith(("raw:", "raw://")) else url for url in urls]
+    allowed, refused, first_error = [], [], None
     for url in urls:
-        validate_url_destination(url)
-    return urls
+        try:
+            validate_url_destination(url)
+            allowed.append(url)
+        except HTTPException as e:
+            first_error = first_error or e
+            refused.append(CrawlResult(url=url, html="", success=False, error_message=e.detail))
+    if first_error and not allowed:
+        raise first_error
+    return allowed, refused
 
 
 async def handle_crawl_request(
@@ -684,7 +697,7 @@ async def handle_crawl_request(
     peak_mem_mb = start_mem_mb
 
     try:
-        urls = _normalize_and_validate_seeds(urls)
+        urls, refused = _normalize_and_validate_seeds(urls)
         browser_config = BrowserConfig.load(browser_config, provenance=Provenance.UNTRUSTED)
         crawler_config = CrawlerRunConfig.load(crawler_config, provenance=Provenance.UNTRUSTED)
         from egress_broker import enforce_egress
@@ -762,6 +775,7 @@ async def handle_crawl_request(
         # Ensure results is always a list
         if not isinstance(results, list):
             results = [results]
+        results = refused + results
 
         end_mem_mb = _get_memory_mb() # <--- Get memory after
         end_time = time.time()
@@ -886,6 +900,12 @@ async def handle_crawl_request(
             else:
                 await release_crawler(crawler)
 
+async def _prepend_results(first: List[CrawlResult], results_gen: AsyncGenerator) -> AsyncGenerator:
+    for result in first:
+        yield result
+    async for result in results_gen:
+        yield result
+
 async def handle_stream_crawl_request(
     urls: List[str],
     browser_config: dict,
@@ -900,7 +920,7 @@ async def handle_stream_crawl_request(
         # SSRF guard: validate every seed URL's destination before fetching,
         # mirroring handle_crawl_request. The streaming path previously skipped
         # this, leaving /crawl/stream (and /crawl with stream=true) unguarded.
-        urls = _normalize_and_validate_seeds(urls)
+        urls, refused = _normalize_and_validate_seeds(urls)
         browser_config = BrowserConfig.load(
             browser_config, provenance=Provenance.UNTRUSTED
         )
@@ -918,12 +938,12 @@ async def handle_stream_crawl_request(
         crawler_config.stream = True
 
         # Deep crawl streaming supports exactly one start URL
-        if crawler_config.deep_crawl_strategy is not None and len(urls) != 1:
+        if crawler_config.deep_crawl_strategy is not None and len(urls) + len(refused) != 1:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
                 detail=(
                     "Deep crawling with stream currently supports exactly one URL per request. "
-                    f"Received {len(urls)} URLs."
+                    f"Received {len(urls) + len(refused)} URLs."
                 ),
             )
 
@@ -968,6 +988,9 @@ async def handle_stream_crawl_request(
                 dispatcher=dispatcher
             )
 
+        if refused:
+            results_gen = _prepend_results(refused, results_gen)
+
         return crawler, results_gen, hooks_info
 
     except (UntrustedConfigError, HookValidationError) as e:
 
@@ -0,0 +1,81 @@
+"""#2288: one refused seed must not sink the whole /crawl or /crawl/stream batch."""
+import json
+
+import pytest
+from fastapi import HTTPException
+
+from crawl4ai import CrawlResult
+
+GOOD = "https://example.com/"
+BAD = "https://no-such-host.example/"
+
+
+@pytest.fixture
+def crawled(monkeypatch):
+    """Refuse BAD, fake the crawler, and record which URLs reached it."""
+    import api
+    import crawler_pool
+
+    seen = []
+
+    def validate(url):
+        if url == BAD:
+            raise HTTPException(400, "URL blocked (SSRF protection): URL blocked")
+
+    class FakeCrawler:
+        async def arun(self, url, **kwargs):
+            seen.append(url)
+            return CrawlResult(url=url, html="ok", success=True)
+
+        async def arun_many(self, urls, config=None, **kwargs):
+            seen.extend(urls)
+            results = [CrawlResult(url=u, html="ok", success=True) for u in urls]
+            if not config.stream:
+                return results
+
+            async def gen():
+                for r in results:
+                    yield r
+            return gen()
+
+    async def get_crawler(*args, **kwargs):
+        return FakeCrawler()
+
+    async def release_crawler(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(api, "validate_url_destination", validate)
+    monkeypatch.setattr(crawler_pool, "get_crawler", get_crawler)
+    monkeypatch.setattr(crawler_pool, "release_crawler", release_crawler)
+    return seen
+
+
+def _auth():
+    from auth import create_access_token
+    return {"Authorization": f"Bearer {create_access_token({'sub': 'u@x.com'})}"}
+
+
+def test_batch_crawls_allowed_seeds_and_reports_refused(stock_client, crawled):
+    r = stock_client.post("/crawl", json={"urls": [GOOD, BAD]}, headers=_auth())
+    assert r.status_code == 200, r.text
+    by_url = {res["url"]: res for res in r.json()["results"]}
+    assert by_url[GOOD]["success"] is True
+    assert by_url[BAD]["success"] is False
+    assert by_url[BAD]["error_message"] == "URL blocked (SSRF protection): URL blocked"
+    assert crawled == [GOOD]
+
+
+def test_stream_crawls_allowed_seeds_and_reports_refused(stock_client, crawled):
+    r = stock_client.post("/crawl/stream", json={"urls": [GOOD, BAD, GOOD]}, headers=_auth())
+    assert r.status_code == 200, r.text
+    lines = [json.loads(line) for line in r.text.splitlines() if line]
+    results = [l for l in lines if "url" in l]
+    assert [(l["url"], l["success"]) for l in results] == [(BAD, False), (GOOD, True), (GOOD, True)]
+    assert crawled == [GOOD, GOOD]
+
+
+@pytest.mark.parametrize("path", ["/crawl", "/crawl/stream"])
+def test_all_refused_is_still_400(stock_client, crawled, path):
+    r = stock_client.post(path, json={"urls": [BAD, BAD]}, headers=_auth())
+    assert r.status_code == 400
+    assert crawled == []