diff --git a/api/ApplyTrack.Api/ApplyTrack.Api.csproj b/api/ApplyTrack.Api/ApplyTrack.Api.csproj
index b56b971..4cb1973 100644
--- a/api/ApplyTrack.Api/ApplyTrack.Api.csproj
+++ b/api/ApplyTrack.Api/ApplyTrack.Api.csproj
@@ -5,7 +5,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <RootNamespace>ApplyTrack.Api</RootNamespace>
-    <Version>1.7.6</Version>
+    <Version>1.7.7</Version>
     <Authors>Aaron K. Clark</Authors>
     <Copyright>Copyright 2026 Aaron K. Clark</Copyright>
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
diff --git a/pyproject.toml b/pyproject.toml
index ba4bdb8..878787e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "applytrack-poller"
-version = "1.7.6"
+version = "1.7.7"
 description = "Discovery poller for OSApplyTrack — fetches and scores remote job leads into shared Postgres."
 requires-python = ">=3.10"
 license = { text = "Apache-2.0" }
diff --git a/src/applytrack/db.py b/src/applytrack/db.py
index de215c9..66ab533 100644
--- a/src/applytrack/db.py
+++ b/src/applytrack/db.py
@@ -30,12 +30,17 @@
 from applytrack.store import AppFields, filename_for
 
 # A plain INSERT (not the importer's upsert): a slug already present is a genuine
-# collision to skip, never an overwrite of a row the user may have since edited.
+# collision, so :meth:`PollRepo.add_lead` retries under a suffixed name rather than
+# overwriting a row the user may have since edited.
 _INSERT_SQL = f"""
 INSERT INTO applications (tenant_id, name, {", ".join(_FIELD_COLUMNS)})
 VALUES (%(tenant_id)s, %(name)s, {", ".join(f"%({c})s" for c in _FIELD_COLUMNS)})
 """
 
+# How many slug names (``stem.md``, ``stem-2.md`` … ``stem-N.md``) ``add_lead`` will
+# try before giving up — a backstop against an unforeseen unique-violation loop.
+_MAX_NAME_ATTEMPTS = 50
+
 # Column order mirrors search_profiles; the keys match Criteria.from_dict's dict.
 _PROFILE_COLUMNS = (
     "keywords", "default_lane", "min_fit_score", "remote_only",
@@ -126,10 +131,25 @@ def mark_seen(self, url_key: str, slug_key: str) -> None:
     def add_lead(self, fields: AppFields) -> str:
         """Stage one new lead, returning its slug ``name``.
 
-        Raises :class:`psycopg.errors.UniqueViolation` if the slug already exists
-        for this tenant; the caller treats that as "skip this listing".
+        When the natural ``company-role`` slug already exists for this tenant —
+        two genuinely distinct postings (distinct URLs) whose titles normalize to
+        the same name — a numeric ``-N`` suffix is appended so the second is
+        staged rather than dropped. Autocommit means each failed INSERT stands
+        alone, so the retry is clean. Only after :data:`_MAX_NAME_ATTEMPTS` free
+        names are exhausted does the underlying
+        :class:`psycopg.errors.UniqueViolation` propagate.
         """
-        name = filename_for(fields)
-        with self._conn.cursor() as cur:
-            cur.execute(_INSERT_SQL, row_params(name, fields, self._t))
-        return name
+        base = filename_for(fields)
+        stem = base[:-3] if base.endswith(".md") else base
+        name = base
+        attempt = 1
+        while True:
+            try:
+                with self._conn.cursor() as cur:
+                    cur.execute(_INSERT_SQL, row_params(name, fields, self._t))
+                return name
+            except psycopg.errors.UniqueViolation:
+                attempt += 1
+                if attempt > _MAX_NAME_ATTEMPTS:
+                    raise
+                name = f"{stem}-{attempt}.md"
diff --git a/src/applytrack/poll.py b/src/applytrack/poll.py
index 526dd3b..16dd079 100644
--- a/src/applytrack/poll.py
+++ b/src/applytrack/poll.py
@@ -16,12 +16,15 @@
 This runtime no longer touches the disk store; it shares one Postgres with the
 API and is driven against a tenant-scoped repo (see :class:`LeadRepo`).
 
-Deduplication is load-bearing — a company must never be pinged twice. For each
-run an in-memory ledger is seeded from the tenant's existing ``applications``
-rows, and every listing is checked against it using two independent keys: the
-normalized listing URL and the normalized ``company + role`` slug. If *either*
-key has been seen, the listing is skipped. New keys are recorded even when
-staging fails, so a transient error never causes a re-ping.
+Deduplication is load-bearing — the same posting must never be staged twice. For
+each run an in-memory ledger is seeded from the tenant's existing ``applications``
+rows. The posting URL is the authoritative key: a listing with a *distinct* URL is
+a distinct lead, even when its ``company + role`` slug matches one already seen
+(e.g. one role posted per city). The slug is only the fallback key for the
+minority of sources that give no URL. Keys are recorded for any listing we
+evaluate — staged, filtered, or link-dead — so we don't re-process it; the lone
+exception is a genuine *write* failure, which leaves the listing unrecorded so a
+later run can retry it.
 """
 
 from __future__ import annotations
@@ -51,7 +54,6 @@
 _TAG_RE = re.compile(r"<[^>]+>")
 _WS_RE = re.compile(r"\s+")
 _SLUG_RE = re.compile(r"[^a-z0-9]+")
-_PARENS_RE = re.compile(r"\([^)]*\)")
 _URL_RE = re.compile(r"https?://[^\s<>\"')]+")
 
 # Signals that a role is remote-friendly, for the optional remote-only filter.
@@ -96,12 +98,12 @@ def _norm_url(url: str) -> str:
 def _norm_slug(company: str, role: str) -> str:
     """Normalized ``company + role`` key, robust to spacing/punctuation.
 
-    Trailing parentheticals are dropped before keying, so the same role posted
-    across locations collapses to one key and a company is not pinged once per
-    city. This errs toward over-dedup, the safe direction for not annoying
-    employers.
+    The *full* role is kept (parentheticals included), so per-city variants like
+    ``Engineer (Campinas)`` vs ``Engineer (São Paulo)`` are distinct keys — they
+    are distinct openings. This is the *fallback* dedup key, used only for
+    listings that carry no URL; when a URL is present it is authoritative (see
+    :meth:`Seen.has`), so two distinct URLs are never collapsed by this slug.
     """
-    role = _PARENS_RE.sub(" ", role)
     return _SLUG_RE.sub("-", f"{company} {role}".lower()).strip("-")
 
 
@@ -177,8 +179,17 @@ class Seen:
     sink: Callable[[str, str], None] | None = None
 
     def has(self, url: str, slug: str) -> bool:
+        """Have we already processed this listing?
+
+        The URL is authoritative: when the listing has one, the verdict is purely
+        whether *that* URL was seen — a distinct URL is a distinct lead, never
+        suppressed by a colliding ``company + role`` slug. The slug is consulted
+        only for URL-less listings, where it is all we have.
+        """
         nu = _norm_url(url)
-        return (bool(nu) and nu in self.urls) or (bool(slug) and slug in self.slugs)
+        if nu:
+            return nu in self.urls
+        return bool(slug) and slug in self.slugs
 
     def note(self, url: str, slug: str) -> tuple[str, str]:
         """Record keys in memory only; return the normalized keys newly added."""
@@ -619,13 +630,26 @@ def fetch_lever(client: httpx.Client, limit: int, slug: str) -> list[Listing]:
 
 
 def make_ats_fetcher(board: AtsBoard) -> Fetcher | None:
-    """Bind a company slug to its provider's fetcher, yielding a plain Fetcher."""
+    """Bind a company slug to its provider's fetcher, yielding a plain Fetcher.
+
+    The returned callable carries a ``provider:slug`` ``__name__`` so a failure of
+    one board is named in the poll diagnostics rather than logged as ``<lambda>``.
+    """
     slug = board.slug
     if board.provider == "greenhouse":
-        return lambda client, limit: fetch_greenhouse(client, limit, slug)
-    if board.provider == "lever":
-        return lambda client, limit: fetch_lever(client, limit, slug)
-    return None
+
+        def _fetch(client: httpx.Client, limit: int) -> list[Listing]:
+            return fetch_greenhouse(client, limit, slug)
+
+    elif board.provider == "lever":
+
+        def _fetch(client: httpx.Client, limit: int) -> list[Listing]:
+            return fetch_lever(client, limit, slug)
+
+    else:
+        return None
+    _fetch.__name__ = f"{board.provider}:{slug}"
+    return _fetch
 
 
 def build_fetchers(criteria: Criteria) -> list[Fetcher]:
@@ -643,14 +667,21 @@ def build_fetchers(criteria: Criteria) -> list[Fetcher]:
 
 
 def _gather(fetchers: Iterable[Fetcher], limit: int) -> list[Listing]:
-    """Run each fetcher, tolerating per-source failures (offline, 4xx, parse)."""
+    """Run each fetcher, tolerating per-source failures (offline, 4xx, parse).
+
+    A failing source is skipped but logged at WARNING with its name, so a
+    dead/renamed source is visible in the worker logs instead of silently looking
+    like an empty job market. Any exception is caught — one broken source (e.g. a
+    changed upstream schema) must never abort the whole run.
+    """
     listings: list[Listing] = []
     with httpx.Client(timeout=20.0, follow_redirects=True, headers=BROWSER_HEADERS) as client:
         for fetch in fetchers:
+            label = getattr(fetch, "__name__", repr(fetch)).removeprefix("fetch_")
             try:
                 listings.extend(fetch(client, limit))
-            except (httpx.HTTPError, ValueError, KeyError):
-                continue
+            except Exception:  # noqa: BLE001 - one bad source must not abort the run
+                logger.warning("poll source %s failed", label, exc_info=True)
     return listings
 
 
@@ -743,14 +774,17 @@ def score_and_stage(
             slug = _norm_slug(item.company, item.role)
             if seen.has(item.link, slug):
                 continue
-            # Record keys up front so a later failure still can't cause a re-ping.
-            seen.add(item.link, slug)
 
+            # We act on this listing one way or another below, so it is recorded
+            # as seen *except* on a genuine write failure (handled at stage time),
+            # which leaves it for a later run to retry.
             if not _passes_location(item, profile):
+                seen.add(item.link, slug)
                 continue
 
             score, hits = classify(item.role, item.description, profile.keywords)
             if not hits or score < profile.min_fit_score:
+                seen.add(item.link, slug)
                 continue
 
             # Block dead postings: don't create an entry we can't actually open.
@@ -759,20 +793,26 @@ def score_and_stage(
                 and item.link
                 and not is_reachable(item.link, client=verify_client)
             ):
+                seen.add(item.link, slug)
                 continue
 
             fields = _to_fields(item, profile.default_lane, score, hits)
             try:
+                # add_lead suffixes -N on a slug-name collision, so a distinct URL
+                # whose name matches an existing row is staged, not dropped.
                 name = repo.add_lead(fields)
             except psycopg.errors.UniqueViolation:
-                logger.debug(
-                    "skipping duplicate lead for %s: %s",
+                # add_lead exhausted its name attempts: surface and leave unseen
+                # so a later run retries it.
+                logger.warning(
+                    "stage failed (slug collision) for %s — %s",
                     item.company,
                     item.role,
                     exc_info=True,
                 )
                 continue
-            # Leads stage as-is; the cover letter is drafted on demand later.
+            # Staged: record keys, then collect. Cover letter is drafted on demand.
+            seen.add(item.link, slug)
             added.append(name)
     finally:
         if verify_client is not None:
diff --git a/src/applytrack/worker.py b/src/applytrack/worker.py
index 8a9a7ed..09647f9 100644
--- a/src/applytrack/worker.py
+++ b/src/applytrack/worker.py
@@ -19,6 +19,7 @@
 
 from __future__ import annotations
 
+import logging
 from collections.abc import Callable, Iterable
 from typing import Protocol
 
@@ -37,6 +38,8 @@
     score_and_stage,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class TenantRepo(LeadRepo, Protocol):
     """A :class:`~applytrack.poll.LeadRepo` that can also load its own profile.
@@ -69,7 +72,9 @@ def _gather_by_source(
 
     Built-in sources key by name (``"remotive"``); ATS boards key by
     ``"{provider}:{slug}"`` — the same keys :func:`_select_for_profile` routes by.
-    Per-source failures yield an empty bucket rather than aborting the gather.
+    A failing source yields an empty bucket and is logged at WARNING (rather than
+    aborting the gather or vanishing silently), so a dead/renamed source is visible
+    in the logs instead of looking like an empty market for every tenant.
     """
     builtin: set[str] = set()
     boards: dict[str, AtsBoard] = {}
@@ -85,16 +90,18 @@ def _gather_by_source(
         for name in sorted(builtin):
             try:
                 gathered[name] = SOURCE_FETCHERS[name](client, limit)
-            except (httpx.HTTPError, ValueError, KeyError):
+            except Exception:  # noqa: BLE001 - one bad source must not abort the gather
                 gathered[name] = []
+                logger.warning("poll source %s failed", name, exc_info=True)
         for key, board in boards.items():
             fetcher = make_ats_fetcher(board)
             if fetcher is None:
                 continue
             try:
                 gathered[key] = fetcher(client, limit)
-            except (httpx.HTTPError, ValueError, KeyError):
+            except Exception:  # noqa: BLE001 - one bad source must not abort the gather
                 gathered[key] = []
+                logger.warning("poll source %s failed", key, exc_info=True)
     return gathered
 
 
@@ -145,6 +152,7 @@ def repo_for(tid: int) -> TenantRepo:
             repos[tid] = repo
         except Exception:  # noqa: BLE001 - one tenant's failure must not abort the rest
             results[tid] = []
+            logger.warning("poll setup failed for tenant %s", tid, exc_info=True)
 
     if gathered is None:
         gathered = _gather_by_source(profiles.values(), limit_per_source)
@@ -159,6 +167,7 @@ def repo_for(tid: int) -> TenantRepo:
             )
         except Exception:  # noqa: BLE001 - one tenant's failure must not abort the rest
             results[tid] = []
+            logger.warning("poll failed for tenant %s", tid, exc_info=True)
     return results
 
 
diff --git a/tests/test_poll.py b/tests/test_poll.py
index 4ca394e..26544dd 100644
--- a/tests/test_poll.py
+++ b/tests/test_poll.py
@@ -4,12 +4,16 @@
 
 from __future__ import annotations
 
+import logging
+
+import httpx
 import psycopg
 import pytest
 
 from applytrack.criteria import AtsBoard, Criteria
 from applytrack.poll import (
     Listing,
+    _gather,
     _looks_remote,
     _passes_location,
     build_fetchers,
@@ -49,6 +53,7 @@ def __init__(
             (self._seen_urls if kind == "url" else self._seen_slugs).add(key)
         self._profile = profile if profile is not None else Criteria()
         self.added: list[AppFields] = []
+        self._names: set[str] = set()
 
     def load_profile(self) -> Criteria:
         return self._profile
@@ -69,9 +74,16 @@ def mark_seen(self, url_key: str, slug_key: str) -> None:
             self._seen_slugs.add(slug_key)
 
     def add_lead(self, fields: AppFields) -> str:
-        name = filename_for(fields)
-        if any(filename_for(f) == name for f in self.added):
-            raise ValueError(f"an application named {name!r} already exists")
+        # Mirror PollRepo.add_lead: suffix -N on a slug-name collision so two
+        # genuinely distinct postings can coexist rather than the second failing.
+        base = filename_for(fields)
+        stem = base[:-3] if base.endswith(".md") else base
+        name = base
+        n = 1
+        while name in self._names:
+            n += 1
+            name = f"{stem}-{n}.md"
+        self._names.add(name)
         self.added.append(fields)
         return name
 
@@ -163,7 +175,8 @@ def test_run_poll_stages_matches_with_default_lane() -> None:
     assert fields.company == "Acme"
 
 
-def test_run_poll_dedupes_same_role() -> None:
+def test_run_poll_dedupes_same_url() -> None:
+    # The same posting seen twice (identical URL) stages exactly once.
     repo = FakeRepo()
     c = Criteria(keywords=["engineer"])
     added = run_poll(
@@ -171,7 +184,55 @@ def test_run_poll_dedupes_same_role() -> None:
         c,
         listings=[
             _lead("Acme", "Backend Engineer", link="https://acme.co/1"),
-            _lead("Acme", "Backend Engineer (Remote)", link="https://acme.co/2"),
+            _lead("Acme", "Backend Engineer", link="https://acme.co/1"),
+        ],
+    )
+    assert len(added) == 1
+
+
+def test_run_poll_keeps_distinct_urls_for_role_variants() -> None:
+    # Per-city variants of one role have distinct URLs -> distinct leads. The old
+    # behavior stripped the parenthetical, collapsed them, and dropped all but one.
+    repo = FakeRepo()
+    c = Criteria(keywords=["engineer"])
+    added = run_poll(
+        repo,
+        c,
+        listings=[
+            _lead("Acme", "Backend Engineer (Campinas)", link="https://acme.co/1"),
+            _lead("Acme", "Backend Engineer (São Paulo)", link="https://acme.co/2"),
+        ],
+    )
+    assert len(added) == 2
+
+
+def test_score_and_stage_unique_names_for_same_title_distinct_urls() -> None:
+    # Same role title (location only in a separate field) -> same slug stem but
+    # distinct URLs. Each must get its own row via a -N suffix, not be dropped.
+    repo = FakeRepo()
+    c = Criteria(keywords=["engineer"])
+    added = score_and_stage(
+        repo,
+        c,
+        listings=[
+            _lead("Acme", "Backend Engineer", link="https://acme.co/1", location="NYC"),
+            _lead("Acme", "Backend Engineer", link="https://acme.co/2", location="SF"),
+        ],
+    )
+    assert len(added) == 2
+    assert added == ["acme-backend-engineer.md", "acme-backend-engineer-2.md"]
+
+
+def test_run_poll_dedupes_urlless_by_slug() -> None:
+    # With no URL, the slug is the only key, so identical url-less posts collapse.
+    repo = FakeRepo()
+    c = Criteria(keywords=["engineer"])
+    added = run_poll(
+        repo,
+        c,
+        listings=[
+            _lead("Acme", "Backend Engineer", link=""),
+            _lead("Acme", "Backend Engineer", link=""),
         ],
     )
     assert len(added) == 1
@@ -250,6 +311,27 @@ def test_run_poll_applies_location_filter() -> None:
     assert added == []
 
 
+def test_gather_logs_and_continues_on_source_failure(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # A failing source is skipped but logged by name; the others still contribute.
+    def boom(client: httpx.Client, limit: int) -> list[Listing]:
+        raise ValueError("upstream schema changed")
+
+    boom.__name__ = "fetch_boom"
+
+    def ok(client: httpx.Client, limit: int) -> list[Listing]:
+        return [_lead("Acme", "Engineer", link="https://acme.co/1")]
+
+    ok.__name__ = "fetch_ok"
+
+    with caplog.at_level(logging.WARNING):
+        listings = _gather([boom, ok], 10)
+
+    assert [item.company for item in listings] == ["Acme"]
+    assert any("boom" in r.getMessage() for r in caplog.records)
+
+
 def test_build_fetchers_includes_sources_and_boards() -> None:
     c = Criteria(ats_boards=[AtsBoard(provider="greenhouse", slug="stripe")])
     fetchers = build_fetchers(c)
@@ -270,9 +352,9 @@ def test_run_poll_dedupes_across_runs_via_seen_ledger() -> None:
     assert len(repo.added) == 1
 
 
-def test_run_poll_honors_preloaded_seen_keys() -> None:
-    # A slug already in the seen table (e.g. from a since-deleted lead) is skipped.
-    repo = FakeRepo(seen=[("slug", "acme-backend-engineer")])
+def test_run_poll_honors_preloaded_seen_url() -> None:
+    # A URL already in the seen table (e.g. from a since-deleted lead) is skipped.
+    repo = FakeRepo(seen=[("url", "acme.co/1")])
     c = Criteria(keywords=["engineer"])
     added = run_poll(
         repo, c, listings=[_lead("Acme", "Backend Engineer", link="https://acme.co/1")]
@@ -281,6 +363,21 @@ def test_run_poll_honors_preloaded_seen_keys() -> None:
     assert repo.added == []
 
 
+def test_run_poll_preloaded_slug_blocks_only_urlless() -> None:
+    # The slug is the fallback key: a preloaded slug blocks a url-less listing, but
+    # a listing carrying a distinct URL is judged by URL and still staged.
+    c = Criteria(keywords=["engineer"])
+
+    urlless = FakeRepo(seen=[("slug", "acme-backend-engineer")])
+    assert run_poll(urlless, c, listings=[_lead("Acme", "Backend Engineer", link="")]) == []
+
+    with_url = FakeRepo(seen=[("slug", "acme-backend-engineer")])
+    added = run_poll(
+        with_url, c, listings=[_lead("Acme", "Backend Engineer", link="https://acme.co/9")]
+    )
+    assert len(added) == 1
+
+
 def test_run_all_tenants_routes_sources_per_profile() -> None:
     # Two tenants enable different sources; each must only ever see roles from the
     # sources it turned on, from one shared per-source gather.