fix: harden Apec fetch behavior

src/job_research/apec/adapter.py

@@ -8,12 +8,15 @@ from urllib.parse import parse_qsl, urlencode, urlparse, urlsplit, urlunsplit
 from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
 from playwright.sync_api import sync_playwright
 
+from job_research.models import ListingError
+
 
 _SEARCH_URL = "https://www.apec.fr/candidat/recherche-emploi.html/emploi"
 _FRANCE_LOCATION_ID = "799"
 _CDI_CONTRACT_ID = "101888"
 _SEARCH_INPUT_SELECTOR = 'input[name="keywords"]'
 _RESULT_LINK_SELECTOR = "a[href*='/candidat/recherche-emploi.html/emploi/detail-offre/']"
+_ZERO_RESULTS_URL_FRAGMENT = "/recherche-avancee"
 _DETAIL_JOB_ID_PATTERN = re.compile(r"/detail-offre/([^/?#]+)")
 _APEC_HOSTS = {"apec.fr", "www.apec.fr"}
 _MAX_PAGES_PER_QUERY = 50
@@ -109,13 +112,55 @@ def _is_public_apec_detail_url(url: str) -> bool:
 class ApecAdapter:
     def __init__(self, max_listings: int = 50) -> None:
         self.max_listings = max_listings
+        self.search_errors: list[ListingError] = []
+        self._browser_context = None
+
+    @contextmanager
+    def browser_session(self):
+        if self._browser_context is not None:
+            yield
+            return
+
+        with sync_playwright() as playwright:
+            browser = playwright.chromium.launch(headless=True)
+            browser_context = browser.new_context()
+            self._browser_context = browser_context
+            try:
+                yield
+            finally:
+                self._browser_context = None
+                browser.close()
+
+    @contextmanager
+    def _open_page(self):
+        if self._browser_context is None:
+            with _open_public_page() as page:
+                yield page
+            return
+
+        page = self._browser_context.new_page()
+        page.set_default_timeout(15_000)
+        try:
+            yield page
+        finally:
+            page.close()
+
+    def _record_search_error(self, query: str, search_filters: ApecSearchFilters, message: str) -> None:
+        self.search_errors.append(
+            ListingError(url=_search_url(query, search_filters), stage="search", message=message)
+        )
+
+    @staticmethod
+    def _is_zero_results_page(page) -> bool:
+        return _ZERO_RESULTS_URL_FRAGMENT in page.url and "error=true" in page.url
 
     def search(self, queries: list[str], search_filters: ApecSearchFilters) -> list[ApecSearchResult]:
         results: list[ApecSearchResult] = []
         seen_keys: set[str] = set()
         usable_search_page_seen = False
+        self.search_errors = []
 
-        with _open_public_page() as page:
+        with self._open_page() as page:
             for query in queries:
                 if not query.strip():
                     continue
@@ -124,6 +169,7 @@ class ApecAdapter:
                     break
 
                 if not _goto_and_wait(page, _search_url(query, search_filters)):
+                    self._record_search_error(query, search_filters, "search page navigation failed")
                     continue
 
                 _accept_cookies_if_present(page)
@@ -131,15 +177,21 @@ class ApecAdapter:
                 try:
                     page.wait_for_selector(_SEARCH_INPUT_SELECTOR, timeout=5_000)
                 except PlaywrightTimeoutError:
+                    self._record_search_error(query, search_filters, "search input did not render")
                     continue
 
-                usable_search_page_seen = True
+                if self._is_zero_results_page(page):
+                    usable_search_page_seen = True
+                    continue
 
                 try:
                     page.wait_for_selector(_RESULT_LINK_SELECTOR, timeout=5_000)
                 except PlaywrightTimeoutError:
+                    self._record_search_error(query, search_filters, "search results did not render")
                     continue
 
+                usable_search_page_seen = True
+
                 result_page_url = page.url
                 seen_page_urls: set[str] = {result_page_url}
                 no_progress_pages = 0
@@ -207,7 +259,7 @@ class ApecAdapter:
         if not _is_public_apec_detail_url(url):
             raise ValueError("ApecAdapter only fetches public Apec URLs")
 
-        with _open_public_page() as page:
+        with self._open_page() as page:
             page.goto(url, wait_until="domcontentloaded")
             page.wait_for_function(
                 """

src/job_research/apec/normalize.py

@@ -6,7 +6,7 @@ from datetime import datetime
 from bs4 import BeautifulSoup
 from bs4.element import NavigableString
 
-from job_research.models import ApecListing
+from job_research.models import ApecListing, ListingWarning
 
 
 _PUBLISHED_AT_PATTERN = re.compile(r"Publi[ée]e le (\d{2}/\d{2}/\d{4})")
@@ -72,6 +72,10 @@ def _detail_block_text(soup: BeautifulSoup, label: str) -> str | None:
     return None
 
 
+def _warning(field: str, message: str) -> ListingWarning:
+    return ListingWarning(field=field, message=message)
+
+
 def _extract_published_at(soup: BeautifulSoup) -> str | None:
     card_offer = soup.select_one(".card-offer")
     if card_offer is None:
@@ -142,27 +146,93 @@ def normalize_apec_listing(
     published_at: str | None = None,
 ) -> ApecListing:
     soup = BeautifulSoup(html, "html.parser")
+    warnings: list[ListingWarning] = []
+
+    title = soup.select_one(".container-details-offer h1")
+    if title is None:
+        title = soup.find("h1")
+        if title is not None:
+            warnings.append(_warning("title", "Recovered title from generic h1 fallback"))
+        else:
+            warnings.append(_warning("title", "Title missing from Apec listing"))
 
-    title = soup.select_one(".container-details-offer h1") or soup.find("h1")
     details_offer_list = soup.select_one(".details-offer-list")
 
     location = None
     contract_type = None
     if details_offer_list is not None:
-        location = details_offer_list.select_one("li:nth-of-type(3)")
-        contract_type = _extract_contract_type(details_offer_list)
+        location_item = details_offer_list.select_one("li:nth-of-type(3)")
+        if location_item is not None:
+            location = location_item.get_text(" ", strip=True)
+        else:
+            warnings.append(_warning("location", "Location missing from details-offer list"))
+
+        contract_item = details_offer_list.select_one("li:nth-of-type(2)")
+        if contract_item is None:
+            warnings.append(_warning("contract_type", "Contract type missing from details-offer list"))
+        else:
+            span = contract_item.find("span")
+            if span is not None:
+                contract_type = _clean_text(span.get_text(" ", strip=True))
+            else:
+                match = _CONTRACT_PATTERN.search(contract_item.get_text(" ", strip=True))
+                if match is not None:
+                    contract_type = match.group(1)
+                    warnings.append(_warning("contract_type", "Recovered contract type from text fallback"))
+                else:
+                    warnings.append(_warning("contract_type", "Contract type missing from details-offer list"))
+    else:
+        warnings.append(_warning("location", "Location missing from Apec listing"))
+        warnings.append(_warning("contract_type", "Contract type missing from Apec listing"))
 
     description_text = _detail_block_text(soup, "Descriptif du poste")
+    if description_text is None:
+        warnings.append(_warning("description_text", "Description missing from Apec listing"))
+
+    if source_job_id is not None:
+        normalized_source_job_id = source_job_id
+    else:
+        ref = soup.select_one(".ref-offre")
+        if ref is None:
+            warnings.append(_warning("source_job_id", "Source job id missing from Apec listing"))
+            normalized_source_job_id = None
+        else:
+            ref_source_job_id = _extract_source_job_id(soup, None)
+            if ref_source_job_id is None:
+                warnings.append(_warning("source_job_id", "Source job id missing from ref-offre"))
+            else:
+                warnings.append(_warning("source_job_id", "Recovered source job id from ref-offre fallback"))
+            normalized_source_job_id = ref_source_job_id
+
+    company = soup.select_one(".card-ents .ents-name")
+    if company is None:
+        for selector, warning_message in (
+            (".card-ents-quote", "Recovered company from .card-ents-quote fallback"),
+            (".details-offer-list li:first-of-type", "Recovered company from details-offer-list fallback"),
+        ):
+            company = soup.select_one(selector)
+            if company is not None:
+                warnings.append(_warning("company", warning_message))
+                break
+
+    company_text = _clean_text(company.get_text(" ", strip=True)) if company is not None else None
+    if company_text is None:
+        warnings.append(_warning("company", "Company missing from Apec listing"))
+
+    published_at_value = published_at or _extract_published_at(soup)
+    if published_at_value is None:
+        warnings.append(_warning("published_at", "Published date missing from Apec listing"))
 
     return ApecListing(
         source="apec",
-        source_job_id=_extract_source_job_id(soup, source_job_id),
+        source_job_id=normalized_source_job_id,
         url=url,
         title=_clean_text(title.get_text(" ", strip=True)) if title else None,
-        company=_extract_company(soup, details_offer_list),
-        location=_clean_text(location.get_text(" ", strip=True)) if location else None,
+        company=company_text,
+        location=_clean_text(location) if location else None,
         contract_type=contract_type,
         description_text=description_text or None,
-        published_at=published_at or _extract_published_at(soup),
+        published_at=published_at_value,
         fetched_at=fetched_at,
+        warnings=warnings,
     )

src/job_research/cli.py

@@ -1,3 +1,4 @@
+from contextlib import nullcontext
 from datetime import datetime, timezone
 import re
 from pathlib import Path
@@ -15,7 +16,7 @@ from job_research.apec.query_derivation import derive_apec_queries, derive_apec_
 from job_research.models import ApecRunMeta, ApecSnapshotMeta, CandidateProfileOutput, ListingError
 from job_research.profile.cv_extractor import extract_cv_signals, extract_pdf_text
 from job_research.profile.merge import build_candidate_profile_output
-from job_research.profile.profile_parser import parse_profile_markdown
+from job_research.profile.profile_parser import AuthoredProfile, parse_profile_markdown
 from job_research.storage import apec_run_paths, load_yaml, save_candidate_profile_yaml
 
 app = typer.Typer(help="Build one canonical candidate profile YAML")
@@ -54,6 +55,25 @@ def _load_candidate_profile(profile_path: Path) -> CandidateProfileOutput:
         raise ValueError(f"invalid candidate-profile.yaml at {profile_path}: {exc}") from exc
 
 
+def _load_cv_text(cv: Path) -> str:
+    try:
+        cv_text = extract_pdf_text(cv) if cv.suffix.lower() == ".pdf" else cv.read_text(encoding="utf-8")
+    except Exception as exc:  # pragma: no cover - defensive boundary
+        raise ValueError(f"CV input not readable at {cv}: {exc}") from exc
+
+    if not cv_text.strip():
+        raise ValueError("No readable text found in CV input")
+
+    return cv_text
+
+
+def _load_authored_profile(profile: Path) -> AuthoredProfile:
+    try:
+        return parse_profile_markdown(profile.read_text(encoding="utf-8"))
+    except Exception as exc:  # pragma: no cover - defensive boundary
+        raise ValueError(f"profile markdown invalid at {profile}: {exc}") from exc
+
+
 @app.callback()
 def main_command() -> None:
     pass
@@ -85,11 +105,13 @@ def build_profile(
 ) -> None:
     """Build candidate-profile.yaml from CV and markdown profile."""
 
-    cv_text = extract_pdf_text(cv) if cv.suffix.lower() == ".pdf" else cv.read_text(encoding="utf-8")
-    if not cv_text.strip():
-        raise ValueError("No readable text found in CV input")
+    try:
+        cv_text = _load_cv_text(cv)
+        authored_profile = _load_authored_profile(profile)
+    except ValueError as exc:
+        typer.echo(str(exc), err=True)
+        raise typer.Exit(code=1)
 
-    authored_profile = parse_profile_markdown(profile.read_text(encoding="utf-8"))
     cv_signals = extract_cv_signals(cv_text)
     candidate_profile = build_candidate_profile_output(cv_signals, authored_profile)
 
@@ -137,7 +159,6 @@ def fetch_apec(
     current = _utc_now().astimezone(timezone.utc)
     run_id = current.strftime("%Y-%m-%dT%H-%M-%S-%fZ")
     run_started_at = current.replace(microsecond=0).strftime("%Y-%m-%dT%H:%M:%SZ")
-    fetched_at = run_started_at
 
     adapter = ApecAdapter(max_listings=50)
     try:
@@ -154,51 +175,56 @@ def fetch_apec(
         raise typer.Exit(code=1)
 
     normalized_listings = []
-    listing_errors: list[ListingError] = []
+    listing_errors: list[ListingError] = list(getattr(adapter, "search_errors", []))
     snapshot_metadata: list[ApecSnapshotMeta] = []
     fetched_count = 0
     successful_fetch_count = 0
 
-    for result in search_results:
-        fetched_count += 1
+    browser_session = getattr(adapter, "browser_session", None)
+    session_context = browser_session() if callable(browser_session) else nullcontext()
 
-        try:
-            html = adapter.fetch_listing_html(result.url)
-        except Exception as exc:  # pragma: no cover - defensive boundary
-            listing_errors.append(ListingError(url=result.url, stage="fetch_html", message=str(exc)))
-            continue
+    with session_context:
+        for result in search_results:
+            fetched_count += 1
+            fetched_at = _utc_now().astimezone(timezone.utc).replace(microsecond=0).strftime("%Y-%m-%dT%H:%M:%SZ")
 
-        successful_fetch_count += 1
+            try:
+                html = adapter.fetch_listing_html(result.url)
+            except Exception as exc:  # pragma: no cover - defensive boundary
+                listing_errors.append(ListingError(url=result.url, stage="fetch_html", message=str(exc)))
+                continue
 
-        snapshot_path = paths["snapshots"] / f"{_snapshot_stem(result.url, result.source_job_id)}.html"
-        snapshot_meta = ApecSnapshotMeta(
-            url=result.url,
-            source_job_id=result.source_job_id,
-            snapshot_file=None,
-            fetched_at=fetched_at,
-        )
+            successful_fetch_count += 1
 
-        try:
-            snapshot_path.write_text(html, encoding="utf-8")
-        except Exception as exc:  # pragma: no cover - defensive boundary
-            listing_errors.append(ListingError(url=result.url, stage="snapshot_write", message=str(exc)))
-        else:
-            snapshot_meta.snapshot_file = snapshot_path.name
-
-        snapshot_metadata.append(snapshot_meta)
-
-        try:
-            listing = normalize_apec_listing(
+            snapshot_path = paths["snapshots"] / f"{_snapshot_stem(result.url, result.source_job_id)}.html"
+            snapshot_meta = ApecSnapshotMeta(
                 url=result.url,
-                html=html,
-                fetched_at=fetched_at,
                 source_job_id=result.source_job_id,
+                snapshot_file=None,
+                fetched_at=fetched_at,
             )
-        except Exception as exc:  # pragma: no cover - defensive boundary
-            listing_errors.append(ListingError(url=result.url, stage="normalize", message=str(exc)))
-            continue
 
-        normalized_listings.append(listing)
+            try:
+                snapshot_path.write_text(html, encoding="utf-8")
+            except Exception as exc:  # pragma: no cover - defensive boundary
+                listing_errors.append(ListingError(url=result.url, stage="snapshot_write", message=str(exc)))
+            else:
+                snapshot_meta.snapshot_file = snapshot_path.name
+
+            snapshot_metadata.append(snapshot_meta)
+
+            try:
+                listing = normalize_apec_listing(
+                    url=result.url,
+                    html=html,
+                    fetched_at=fetched_at,
+                    source_job_id=result.source_job_id,
+                )
+            except Exception as exc:  # pragma: no cover - defensive boundary
+                listing_errors.append(ListingError(url=result.url, stage="normalize", message=str(exc)))
+                continue
+
+            normalized_listings.append(listing)
 
     if search_results and successful_fetch_count == 0:
         typer.echo("No listings could be fetched or normalized from Apec", err=True)

tests/apec/test_adapter.py

@@ -48,17 +48,21 @@ class _FakeDetailPage:
         *,
         rendered_html: str = "<html>rendered</html>",
         search_ready: bool = True,
+        zero_result_queries: set[str] | None = None,
     ) -> None:
         self.result_pages = result_pages
         self.rendered_html = rendered_html
         self.shell_html = "<html>shell</html>"
         self.waited_functions: list[tuple[str, int | None]] = []
         self.search_ready = search_ready
+        self.zero_result_queries = zero_result_queries or set()
         self.goto_urls: list[str] = []
         self.current_query = ""
         self.current_page = 0
         self.url = ""
         self.rendered = False
+        self.default_timeout: int | None = None
+        self.closed = False
 
     def goto(self, url: str, wait_until: str | None = None) -> None:
         self.goto_urls.append(url)
@@ -71,12 +75,21 @@ class _FakeDetailPage:
         if "page" in params:
             self.current_page = int(params["page"][0])
 
+        if self.current_query in self.zero_result_queries and "/detail-offre/" not in parsed_url.path:
+            self.url = (
+                f"{parsed_url.scheme}://{parsed_url.netloc}"
+                f"{parsed_url.path}/recherche-avancee?{parsed_url.query}&error=true"
+            )
+
         if "/detail-offre/" in parsed_url.path:
             self.rendered = False
 
     def wait_for_load_state(self, state: str) -> None:
         return None
 
+    def set_default_timeout(self, timeout: int) -> None:
+        self.default_timeout = timeout
+
     def wait_for_function(self, function: str, polling: int | None = None, timeout: int | None = None) -> None:
         self.waited_functions.append((function, polling))
         self.rendered = True
@@ -106,6 +119,9 @@ class _FakeDetailPage:
     def current_results(self) -> list[str]:
         return self.result_pages.get(self.current_query, {}).get(self.current_page, [])
 
+    def close(self) -> None:
+        self.closed = True
+
 
 @contextmanager
 def _fake_open_public_page(page: _FakeDetailPage):
@@ -314,6 +330,39 @@ def test_search_raises_when_every_query_fails_to_load_a_search_page(monkeypatch)
         )
 
 
+def test_search_treats_zero_results_redirect_as_usable_and_records_other_failures(monkeypatch) -> None:
+    page = _FakeDetailPage(
+        {"alpha": {0: []}, "beta": {0: []}},
+        zero_result_queries={"alpha"},
+    )
+    monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
+
+    adapter = ApecAdapter(max_listings=10)
+    results = adapter.search(
+        ["alpha", "beta"],
+        search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
+    )
+
+    assert results == []
+    assert [error.stage for error in adapter.search_errors] == ["search"]
+    assert "beta" in adapter.search_errors[0].url
+
+
+def test_search_raises_when_every_query_renders_broken_search_shell(monkeypatch) -> None:
+    page = _FakeDetailPage({"alpha": {0: []}, "beta": {0: []}}, search_ready=True)
+    monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
+
+    adapter = ApecAdapter(max_listings=10)
+
+    with pytest.raises(adapter_module.ApecSearchError):
+        adapter.search(
+            ["alpha", "beta"],
+            search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
+        )
+
+    assert [error.stage for error in adapter.search_errors] == ["search", "search"]
+
+
 def test_fetch_listing_html_waits_for_rendered_offer_content(monkeypatch) -> None:
     page = _FakeDetailPage({}, rendered_html="<html>rendered offer</html>")
     monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
@@ -351,3 +400,85 @@ def test_fetch_listing_html_rejects_non_apec_hosts() -> None:
 
     with pytest.raises(ValueError):
         adapter.fetch_listing_html("https://evilapec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111")
+
+
+def test_fetch_listing_html_reuses_browser_context_across_calls(monkeypatch) -> None:
+    class FakePage:
+        def __init__(self) -> None:
+            self.goto_urls: list[str] = []
+            self.default_timeout: int | None = None
+
+        def set_default_timeout(self, timeout: int) -> None:
+            self.default_timeout = timeout
+
+        def goto(self, url: str, wait_until: str | None = None) -> None:
+            self.goto_urls.append(url)
+
+        def wait_for_function(self, function: str, polling: int | None = None, timeout: int | None = None) -> None:
+            return None
+
+        def content(self) -> str:
+            return "<html>shared</html>"
+
+        def close(self) -> None:
+            return None
+
+    class FakeBrowserContext:
+        def __init__(self) -> None:
+            self.new_page_calls = 0
+
+        def new_page(self) -> FakePage:
+            self.new_page_calls += 1
+            return FakePage()
+
+        def close(self) -> None:
+            return None
+
+    class FakeBrowser:
+        def __init__(self, browser_context: FakeBrowserContext) -> None:
+            self.browser_context = browser_context
+
+        def new_context(self) -> FakeBrowserContext:
+            return self.browser_context
+
+        def close(self) -> None:
+            return None
+
+    class FakeChromium:
+        def __init__(self, browser: FakeBrowser) -> None:
+            self.browser = browser
+            self.launch_calls = 0
+
+        def launch(self, headless: bool = True) -> FakeBrowser:
+            self.launch_calls += 1
+            return self.browser
+
+    class FakePlaywright:
+        def __init__(self, chromium: FakeChromium) -> None:
+            self.chromium = chromium
+
+    class FakePlaywrightManager:
+        def __init__(self, chromium: FakeChromium) -> None:
+            self.playwright = FakePlaywright(chromium)
+
+        def __enter__(self) -> FakePlaywright:
+            return self.playwright
+
+        def __exit__(self, exc_type, exc, tb) -> None:
+            return None
+
+    browser_context = FakeBrowserContext()
+    browser = FakeBrowser(browser_context)
+    chromium = FakeChromium(browser)
+
+    monkeypatch.setattr(adapter_module, "sync_playwright", lambda: FakePlaywrightManager(chromium))
+
+    adapter = ApecAdapter()
+    with adapter.browser_session():
+        html_one = adapter.fetch_listing_html("https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111")
+        html_two = adapter.fetch_listing_html("https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222")
+
+    assert html_one == "<html>shared</html>"
+    assert html_two == "<html>shared</html>"
+    assert chromium.launch_calls == 1
+    assert browser_context.new_page_calls == 2

tests/apec/test_normalize.py

@@ -124,3 +124,44 @@ def test_normalize_apec_listing_uses_details_offer_list_company_fallback() -> No
 
     assert listing.company == "Fallback Company"
     assert listing.description_text == "Build pipelines"
+
+
+def test_normalize_apec_listing_records_warnings_for_fallback_and_missing_fields() -> None:
+    html = """
+    <html>
+      <body>
+        <main class="container-details-offer">
+          <div class="card-offer">
+            <div class="ref-offre">Ref. Apec : 178554452W</div>
+            <ul class="details-offer-list mb-20">
+              <li>Fallback Company</li>
+              <li>1 CDI</li>
+            </ul>
+          </div>
+          <article class="card card-ents mb-20">
+            <div class="list-hzt mb-20">
+              <span class="card-ents-quote">Fallback Company</span>
+            </div>
+          </article>
+        </main>
+        <h1>Fallback Title</h1>
+      </body>
+    </html>
+    """
+
+    listing = normalize_apec_listing(
+        url="https://example.test/job/123",
+        html=html,
+        fetched_at="2026-06-01T10:00:00Z",
+        source_job_id=None,
+    )
+
+    assert [warning.field for warning in listing.warnings] == [
+        "title",
+        "location",
+        "contract_type",
+        "description_text",
+        "source_job_id",
+        "company",
+        "published_at",
+    ]

tests/test_apec_cli.py

@@ -9,6 +9,7 @@ from typer.testing import CliRunner
 from job_research.apec.adapter import ApecSearchFilters, ApecSearchResult
 from job_research.apec.normalize import normalize_apec_listing as real_normalize_apec_listing
 from job_research.cli import app
+from job_research.models import ListingError
 
 
 def _fixed_now() -> datetime:
@@ -344,6 +345,65 @@ def test_fetch_apec_succeeds_with_empty_search_results(tmp_path, monkeypatch) ->
     }
 
 
+def test_fetch_apec_records_search_failures_as_partial_success(tmp_path, monkeypatch) -> None:
+    data_root = tmp_path / "data"
+    data_root.mkdir()
+    (data_root / "candidate-profile.yaml").write_text(
+        dedent(
+            """
+            target_roles:
+              - Role From YAML
+            """
+        ).strip(),
+        encoding="utf-8",
+    )
+
+    class PartialSearchFailureAdapter:
+        instances: list["PartialSearchFailureAdapter"] = []
+
+        def __init__(self, max_listings: int = 50) -> None:
+            self.max_listings = max_listings
+            self.search_calls: list[list[str]] = []
+            self.fetch_calls: list[str] = []
+            self.search_errors = [
+                ListingError(
+                    url="https://www.apec.fr/candidat/recherche-emploi.html/emploi?motsCles=Role+From+YAML&lieux=799&typesContrat=101888&page=0",
+                    stage="search",
+                    message="search results did not render",
+                )
+            ]
+            PartialSearchFailureAdapter.instances.append(self)
+
+        def search(self, queries: list[str], search_filters=None) -> list[ApecSearchResult]:
+            self.search_calls.append(list(queries))
+            return []
+
+        def fetch_listing_html(self, url: str) -> str:
+            self.fetch_calls.append(url)
+            raise AssertionError("fetch should not be called when search returns no results")
+
+    monkeypatch.setattr("job_research.cli.ApecAdapter", PartialSearchFailureAdapter)
+    monkeypatch.setattr("job_research.cli._utc_now", _fixed_now)
+
+    result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)])
+
+    assert result.exit_code == 0
+    assert "fetched=0" in result.stdout
+    assert "normalized=0" in result.stdout
+    assert "deduplicated=0" in result.stdout
+    assert "failed=1" in result.stdout
+
+    run_dir = data_root / "apec" / "runs" / FIXED_RUN_ID
+    run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8"))
+    assert run_meta_payload["listing_errors"] == [
+        {
+            "url": "https://www.apec.fr/candidat/recherche-emploi.html/emploi?motsCles=Role+From+YAML&lieux=799&typesContrat=101888&page=0",
+            "stage": "search",
+            "message": "search results did not render",
+        }
+    ]
+
+
 def test_fetch_apec_uses_unique_run_dirs_for_runs_started_within_same_second(
     tmp_path,
     monkeypatch,
@@ -576,6 +636,72 @@ def test_fetch_apec_reads_profile_and_writes_run_artifacts(tmp_path, monkeypatch
     }
 
 
+def test_fetch_apec_uses_per_listing_fetch_timestamps(tmp_path, monkeypatch) -> None:
+    data_root = tmp_path / "data"
+    data_root.mkdir()
+    (data_root / "candidate-profile.yaml").write_text(
+        dedent(
+            """
+            target_roles:
+              - Role From YAML
+            """
+        ).strip(),
+        encoding="utf-8",
+    )
+
+    html = _apec_detail_html()
+
+    class TimestampedApecAdapter:
+        instances: list["TimestampedApecAdapter"] = []
+
+        def __init__(self, max_listings: int = 50) -> None:
+            self.max_listings = max_listings
+            self.search_calls: list[list[str]] = []
+            self.fetch_calls: list[str] = []
+            TimestampedApecAdapter.instances.append(self)
+
+        def search(self, queries: list[str], search_filters=None) -> list[ApecSearchResult]:
+            self.search_calls.append(list(queries))
+            return [
+                ApecSearchResult(url="https://example.test/job/123", source_job_id="job-123"),
+                ApecSearchResult(url="https://example.test/job/456", source_job_id="job-456"),
+            ]
+
+        def fetch_listing_html(self, url: str) -> str:
+            self.fetch_calls.append(url)
+            return html
+
+    timestamps = [
+        datetime(2026, 6, 1, 10, 0, 0, 111111, tzinfo=timezone.utc),
+        datetime(2026, 6, 1, 10, 0, 1, 222222, tzinfo=timezone.utc),
+        datetime(2026, 6, 1, 10, 0, 2, 333333, tzinfo=timezone.utc),
+    ]
+
+    def next_now() -> datetime:
+        return timestamps.pop(0)
+
+    monkeypatch.setattr("job_research.cli.ApecAdapter", TimestampedApecAdapter)
+    monkeypatch.setattr("job_research.cli._utc_now", next_now)
+
+    result = CliRunner().invoke(app, ["fetch-apec", "--data-root", str(data_root)])
+
+    assert result.exit_code == 0
+
+    run_dir = data_root / "apec" / "runs" / "2026-06-01T10-00-00-111111Z"
+    listings_payload = yaml.safe_load((run_dir / "listings.yaml").read_text(encoding="utf-8"))
+    assert [item["fetched_at"] for item in listings_payload] == [
+        "2026-06-01T10:00:01Z",
+        "2026-06-01T10:00:02Z",
+    ]
+
+    run_meta_payload = yaml.safe_load((run_dir / "run-meta.yaml").read_text(encoding="utf-8"))
+    assert run_meta_payload["run_started_at"] == "2026-06-01T10:00:00Z"
+    assert [item["fetched_at"] for item in run_meta_payload["snapshots"]] == [
+        "2026-06-01T10:00:01Z",
+        "2026-06-01T10:00:02Z",
+    ]
+
+
 def test_fetch_apec_records_partial_failures_without_losing_successful_listings(
     tmp_path,
     monkeypatch,

tests/test_cli.py

@@ -1,6 +1,9 @@
 from subprocess import run
 from textwrap import dedent
 
+from typer.testing import CliRunner
+
+from job_research.cli import app
 from job_research.storage import load_yaml
 
 
@@ -189,3 +192,100 @@ def test_build_profile_rejects_empty_cv_text_before_writing(tmp_path) -> None:
     assert result.returncode != 0
     assert not out.exists()
     assert "No readable text found in CV input" in result.stderr
+    assert "Traceback" not in result.stderr
+
+
+def test_build_profile_reports_unreadable_pdf_input_cleanly(tmp_path, monkeypatch) -> None:
+    cv = tmp_path / "cv.pdf"
+    cv.write_bytes(b"%PDF-1.4\n")
+    profile = tmp_path / "profile.md"
+    profile.write_text(
+        dedent(
+            """
+            # Candidate Profile
+
+            ## Summary
+            Junior data engineer.
+
+            ## Target Roles
+            - Data Engineer
+
+            ## Strengths
+            - Python
+
+            ## Skills To Emphasize
+            - BigQuery
+
+            ## Constraints
+            - CDI only
+
+            ## Notes
+            - Slight preference for French listings.
+            """
+        ).strip(),
+        encoding="utf-8",
+    )
+    out = tmp_path / "candidate-profile.yaml"
+
+    def broken_extract_pdf_text(path):
+        raise ValueError("broken pdf")
+
+    monkeypatch.setattr("job_research.cli.extract_pdf_text", broken_extract_pdf_text)
+
+    result = CliRunner().invoke(app, ["build-profile", "--cv", str(cv), "--profile", str(profile), "--out", str(out)])
+
+    assert result.exit_code == 1
+    assert not out.exists()
+    assert "CV input not readable" in result.stderr
+    assert "broken pdf" in result.stderr
+    assert "Traceback" not in result.stderr
+
+
+def test_build_profile_reports_malformed_profile_markdown_cleanly(tmp_path) -> None:
+    cv = tmp_path / "cv.txt"
+    cv.write_text(
+        dedent(
+            """
+            Tonio Example
+            Location: France
+            Languages: French, English
+            Skills: Python
+            """
+        ).strip(),
+        encoding="utf-8",
+    )
+    profile = tmp_path / "profile.md"
+    profile.write_text(
+        dedent(
+            """
+            # Candidate Profile
+
+            ## Summary
+            Junior data engineer.
+
+            ## Target Roles
+            Data Engineer
+
+            ## Strengths
+            - Python
+
+            ## Skills To Emphasize
+            - BigQuery
+
+            ## Constraints
+            - CDI only
+
+            ## Notes
+            - Slight preference for French listings.
+            """
+        ).strip(),
+        encoding="utf-8",
+    )
+    out = tmp_path / "candidate-profile.yaml"
+
+    result = CliRunner().invoke(app, ["build-profile", "--cv", str(cv), "--profile", str(profile), "--out", str(out)])
+
+    assert result.exit_code == 1
+    assert not out.exists()
+    assert "profile markdown invalid" in result.stderr
+    assert "Traceback" not in result.stderr