diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2b763293d..fe97944af 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@ Application releases and Alembic schema revisions are separate identifiers.
 ### Fixes
 
 - Reduce delays in automated article review as the topic and tag catalog grows.
+- Keep topic edits responsive while completed article analyses prepare publication checks.
 - Keep sign-in codes, state values, and authentication credentials out of request logs and browser diagnostics.
 
 ## 0.0.50 — 2026-10-03
diff --git a/apps/aggregator/src/analysis_tasks.py b/apps/aggregator/src/analysis_tasks.py
index 5d7a4a614..a2e64a470 100644
--- a/apps/aggregator/src/analysis_tasks.py
+++ b/apps/aggregator/src/analysis_tasks.py
@@ -17,12 +17,14 @@
     fail_analysis,
     refresh_superseded_analysis,
     require_primary_topic,
+    reuse_candidates,
     snapshot_hash,
     source_snapshot,
     validate_evidence,
 )
 from devfeed_core.analysis_wire import compact_request, restore_identities
 from devfeed_core.article_jobs import approved_sources
+from devfeed_core.catalog_cache import reuse_snapshots, snapshot_current
 from devfeed_core.config import get_settings
 from devfeed_core.db import session_factory
 from devfeed_core.inference_validation import feedback_prompt, validation_feedback
@@ -33,6 +35,7 @@
 from devfeed_core.models import Article, ArticleAnalysisJob, ArticleContent, utcnow
 from pydantic import ValidationError
 from sqlalchemy import select
+from sqlalchemy.orm import lazyload
 
 from devfeed_aggregator.analysis_telemetry import record_attempt
 from devfeed_aggregator.codex_client import AnalysisError, CodexClient
@@ -41,6 +44,47 @@
 logger = logging.getLogger(__name__)
 
 
+class _ApplicationInputsChanged(Exception):
+    """Release publication locks before preparing changed catalog/content inputs."""
+
+
+def _prepare_application(factory, article_id, snapshot):
+    with factory() as session:
+        current_catalog = catalog(session)
+        article = session.scalar(
+            select(Article).where(Article.id == article_id).options(lazyload("*"))
+        )
+        current = (
+            source_snapshot(article, session.get(ArticleContent, article_id))
+            if article is not None
+            else None
+        )
+    # Warm both validity checks and any replacement request with no open transaction.
+    analysis_candidates(current_catalog, snapshot)
+    if current is not None and current != snapshot:
+        analysis_candidates(current_catalog, current)
+    return current_catalog, current
+
+
+def _apply_prepared(factory, article_id, snapshot, operation):
+    from devfeed_core.transaction_retry import run_transaction
+
+    for attempt in range(3):
+        # Do not retain full retrieval indexes during the external inference wait.
+        with reuse_snapshots(), reuse_candidates():
+            taxonomy, current = _prepare_application(factory, article_id, snapshot)
+            try:
+                return run_transaction(
+                    factory,
+                    lambda session, taxonomy=taxonomy, current=current: operation(
+                        session, taxonomy, current
+                    ),
+                )
+            except _ApplicationInputsChanged:
+                if attempt == 2:
+                    raise  # Existing durable dependency retries handle continued churn.
+
+
 def analyze_article(job_id: str) -> None:
     with job_log_context("analysis", job_id):
         try:
@@ -108,7 +152,9 @@ def _analyze_claimed(settings, factory, identifier, token, snapshot, article_id)
     try:
         # No transaction remains open while waiting for Codex.
         with factory() as session:
-            taxonomy = analysis_candidates(catalog(session), snapshot)
+            inference_catalog = catalog(session)
+        taxonomy = analysis_candidates(inference_catalog, snapshot)
+        del inference_catalog
         with factory.begin() as session:
             job = owned_job(session, ArticleAnalysisJob, identifier, token)
             if job is None:
@@ -167,7 +213,7 @@ def _analyze_claimed(settings, factory, identifier, token, snapshot, article_id)
                 if article is not None and source_ids and article.review_status != "rejected":
                     propose_source_topics(session, article)
 
-        def apply_result(session):
+        def apply_result(session, current_catalog, prepared_snapshot):
             job = owned_job(session, ArticleAnalysisJob, identifier, token)
             if job is None:
                 logger.warning("article_analysis_lease_lost")
@@ -193,7 +239,12 @@ def apply_result(session):
                     from devfeed_core.topics import lock_topics
 
                     lock_topics(session, read=True)
-                    current_catalog = catalog(session)
+                    if source_snapshot(
+                        article, session.get(ArticleContent, article_id)
+                    ) != prepared_snapshot or not snapshot_current(
+                        session, ("topics", "tags"), current_catalog
+                    ):
+                        raise _ApplicationInputsChanged()
                     if not analysis_catalog_current(job, current_catalog, snapshot):
                         finish_job(job, "superseded", utcnow())
                     else:
@@ -205,12 +256,10 @@ def apply_result(session):
                             apply_publication_policy(
                                 session, article, job, taxonomy=current_catalog
                             )
-                    refresh_superseded_analysis(session, article, job)
+                    refresh_superseded_analysis(session, article, job, taxonomy=current_catalog)
             return job.outcome
 
-        from devfeed_core.transaction_retry import run_transaction
-
-        outcome = run_transaction(factory, apply_result)
+        outcome = _apply_prepared(factory, article_id, snapshot, apply_result)
         logger.info("article_analysis_completed", extra={"outcome": outcome})
     except Exception as exc:
         feedback = validation_feedback(exc)
diff --git a/packages/core/OBSERVABILITY.md b/packages/core/OBSERVABILITY.md
index b0aaf2d9a..c5bf06fed 100644
--- a/packages/core/OBSERVABILITY.md
+++ b/packages/core/OBSERVABILITY.md
@@ -197,6 +197,16 @@ Compare `devfeed_background_duration_seconds` for `scheduler.tick`, scheduler CP
 and topic-lock wait latency over matching windows after rollout. Local catalog
 benchmarks do not establish production tick latency.
 
+Analysis workers also prepare current catalog candidates and identity membership
+before acquiring source, article and topic locks. A revision or source-content
+change after preparation rolls back the application transaction and prepares the
+new inputs outside the locks. Up to three preparations can reuse the same
+inference result; continued churn uses the existing durable dependency retry
+policy. Reanalysis of superseded content receives that validated catalog rather
+than loading and ranking the entire catalog inside the application transaction.
+Candidate scopes begin after inference, so they do not retain a full catalog
+through the external model wait. Evidence and publication guards remain required.
+
 The discovery profile covers feed facets with language, content-type and source
 filters. Broad browsing uses early-exit probes; selective topic, tag and text
 searches resolve their matching articles once. It checks two SQL queries and bounded article visits on a larger fixture:
diff --git a/packages/core/src/analysis.py b/packages/core/src/analysis.py
index c7d214dbf..5bf6eefc3 100644
--- a/packages/core/src/analysis.py
+++ b/packages/core/src/analysis.py
@@ -568,7 +568,11 @@ def validate_evidence(result: Classifications, snapshot: dict, taxonomy: dict) -
         ("topics", result.topics),
         ("tags", result.tags),
     ):
-        valid = {item["id"] for item in taxonomy[field]}
+        valid = (
+            _prepared_catalog(taxonomy).current(field)
+            if _ranking.get() is not None
+            else {item["id"] for item in taxonomy[field]}
+        )
         for item in selections:
             identifier = item.topic_id if isinstance(item, TopicSelection) else item.id
             if str(identifier) not in valid:
@@ -716,7 +720,9 @@ def request_analysis(
     return job
 
 
-def refresh_superseded_analysis(session: Session, article: Article, job: ArticleAnalysisJob):
+def refresh_superseded_analysis(
+    session: Session, article: Article, job: ArticleAnalysisJob, *, taxonomy=None
+):
     """Coalescing an active delivery must not strand newer source evidence."""
     if job.outcome != "superseded" or article.review_status != "pending":
         return None
@@ -726,10 +732,11 @@ def refresh_superseded_analysis(session: Session, article: Article, job: Article
         # not snapshot the candidate hash, so catalog changes cannot be inferred.
         if job.catalog_hash is None or article.editorial_revision != job.editorial_revision:
             return None
-        if analysis_catalog_current(job, catalog(session), current):
+        current_catalog = catalog(session) if taxonomy is None else taxonomy
+        if analysis_catalog_current(job, current_catalog, current):
             return None
     session.flush()  # Release the active-job uniqueness slot before enqueueing.
-    return request_analysis(session, article.id, automatic=True)
+    return request_analysis(session, article.id, automatic=True, taxonomy=taxonomy)
 
 
 def backfill_analyses(
diff --git a/tests/test_analysis.py b/tests/test_analysis.py
index 76ed3be8d..385cbaf44 100644
--- a/tests/test_analysis.py
+++ b/tests/test_analysis.py
@@ -128,6 +128,16 @@ def test_evidence_must_be_from_input_and_ids_from_dynamic_catalog():
         analysis.validate_evidence(invalid, SNAPSHOT, CATALOG)
 
 
+def test_prepared_catalog_membership_rejects_a_removed_identity():
+    with analysis.reuse_candidates():
+        analysis.analysis_candidates(CATALOG, SNAPSHOT)
+        analysis.validate_evidence(result(), SNAPSHOT, CATALOG)
+        changed = {"topics": [], "tags": []}
+        analysis.analysis_candidates(changed, SNAPSHOT)
+        with pytest.raises(ValueError, match="unknown"):
+            analysis.validate_evidence(result(), SNAPSHOT, changed)
+
+
 @pytest.mark.parametrize(
     "values",
     [
@@ -382,7 +392,7 @@ def test_new_content_coalesced_into_active_analysis_gets_a_followup(monkeypatch)
     monkeypatch.setattr(analysis, "request_analysis", lambda *a, **kw: calls.append((a, kw)))
     finish_job(job, "superseded", utcnow())
     analysis.refresh_superseded_analysis(db, current, job)
-    assert len(calls) == 1 and calls[0][1] == {"automatic": True}
+    assert len(calls) == 1 and calls[0][1] == {"automatic": True, "taxonomy": None}
 
 
 @pytest.mark.parametrize("status", ["approved", "rejected", "pending"])
diff --git a/tests/test_analysis_application_locks.py b/tests/test_analysis_application_locks.py
new file mode 100644
index 000000000..a0a481244
--- /dev/null
+++ b/tests/test_analysis_application_locks.py
@@ -0,0 +1,254 @@
+"""Catalog writers can progress during ranking without allowing stale applications."""
+
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
+from threading import Event, get_ident
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+from devfeed_aggregator import analysis_tasks
+from devfeed_core import analysis
+from devfeed_core.db import get_engine
+from devfeed_core.models import Article, ArticleAnalysisJob, utcnow
+from devfeed_core.topics import lock_topics
+from sqlalchemy import event, select, text
+from sqlalchemy.orm import Session
+from test_automation_integration import enable, seed
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def worker_article(database, monkeypatch):
+    enable(monkeypatch)
+    with database.begin() as session:
+        source, article, topic = seed(session, mode="auto")
+        job = analysis.request_analysis(session, article.id)
+        state = SimpleNamespace(
+            source_id=source.id,
+            article_id=article.id,
+            topic_id=topic.id,
+            job_id=job.id,
+            inferences=[],
+        )
+
+    def complete(*args):
+        state.inferences.append(True)
+        return {
+            "outcome": "ready",
+            "developer_relevance": "relevant",
+            "language": "en",
+            "content_type": "tutorial",
+            "content_format": "article",
+            "ai_summary": "A guide to routing in Angular applications.",
+            "ai_title": None,
+            "title_evidence": None,
+            "page_kind": "article",
+            "ai_description": None,
+            "topics": [
+                {
+                    "topic_id": str(state.topic_id),
+                    "role": "primary",
+                    "relevance": 0.9,
+                    "evidence": "Angular",
+                }
+            ],
+            "tags": [],
+            "reasons": [],
+        }
+
+    monkeypatch.setattr(analysis_tasks, "session_factory", lambda: database)
+    monkeypatch.setattr(analysis_tasks, "CodexClient", lambda _: SimpleNamespace(complete=complete))
+    return state
+
+
+@pytest.fixture
+def unlocked_ranking(worker_article, monkeypatch):
+    state = SimpleNamespace(calls=0, hook=None)
+    connections, locked = defaultdict(int), defaultdict(bool)
+    score = analysis._candidate_scores
+
+    def checkout(*args):
+        connections[get_ident()] += 1
+
+    def checkin(*args):
+        connections[get_ident()] -= 1
+
+    def capture(conn, cursor, statement, parameters, context, executemany):
+        if "FOR UPDATE" in statement or "LOCK TABLE topics" in statement:
+            locked[get_ident()] = True
+
+    def released(session, transaction):
+        if transaction.parent is None:
+            locked[get_ident()] = False
+
+    def ranking(*args):
+        assert not locked[get_ident()], "Ranking cannot hold publication or catalog locks"
+        assert connections[get_ident()] == 0, "Ranking cannot keep a database connection"
+        state.calls += 1
+        if state.hook:
+            state.hook(state.calls)
+        return score(*args)
+
+    monkeypatch.setattr(analysis, "_candidate_scores", ranking)
+    listeners = [
+        (get_engine(), "checkout", checkout),
+        (get_engine(), "checkin", checkin),
+        (get_engine(), "before_cursor_execute", capture),
+        (Session, "after_transaction_end", released),
+    ]
+    for target, name, callback in listeners:
+        event.listen(target, name, callback)
+    try:
+        yield state
+    finally:
+        for target, name, callback in listeners:
+            event.remove(target, name, callback)
+
+
+def test_catalog_writer_commits_while_worker_ranking_is_paused(
+    database, worker_article, unlocked_ranking
+):
+    paused, resume = Event(), Event()
+
+    def pause(call):
+        if call == 2:  # Application preparation follows the inference catalog ranking.
+            paused.set()
+            assert resume.wait(10), "The writer did not finish while ranking was paused"
+
+    unlocked_ranking.hook = pause
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        worker = executor.submit(analysis_tasks._analyze, worker_article.job_id)
+        try:
+            assert paused.wait(10), "Application preparation did not reach ranking"
+            with database.begin() as writer:
+                writer.execute(text("SET LOCAL lock_timeout='2s'"))
+                lock_topics(writer)
+                writer.execute(
+                    text(
+                        "UPDATE topics SET description='Catalog maintenance metadata' WHERE id=:id"
+                    ),
+                    {"id": worker_article.topic_id},
+                )
+        finally:
+            resume.set()
+        worker.result(timeout=15)
+    with database() as session:
+        assert session.get(Article, worker_article.article_id).publication_status == "published"
+        assert session.get(ArticleAnalysisJob, worker_article.job_id).outcome == "applied"
+    assert unlocked_ranking.calls == 3  # Revision changed, so prepare again outside locks.
+    assert len(worker_article.inferences) == 1
+
+
+@pytest.mark.parametrize(
+    "change", ["topic", "tag", "content", "editorial", "source", "disabled_source", "lease"]
+)
+def test_changed_inputs_or_ownership_cannot_apply_a_stale_result(
+    database, worker_article, unlocked_ranking, monkeypatch, change
+):
+    prepare = analysis_tasks._prepare_application
+    changed = []
+
+    def racing_prepare(*args):
+        prepared = prepare(*args)
+        if not changed:
+            changed.append(True)
+            with database.begin() as editor:
+                if change == "topic":
+                    editor.execute(
+                        text("UPDATE topics SET name='Changed identity' WHERE id=:id"),
+                        {"id": worker_article.topic_id},
+                    )
+                elif change == "tag":
+                    editor.execute(
+                        text(
+                            "INSERT INTO tags(id,name,slug,aliases) "
+                            "VALUES (gen_random_uuid(),'New label','new-label','{}')"
+                        )
+                    )
+                elif change == "content":
+                    editor.execute(
+                        text(
+                            "UPDATE articles SET summary=summary || ' New source evidence.' "
+                            "WHERE id=:id"
+                        ),
+                        {"id": worker_article.article_id},
+                    )
+                elif change == "editorial":
+                    editor.execute(
+                        text(
+                            "UPDATE articles SET editorial_revision=editorial_revision+1 "
+                            "WHERE id=:id"
+                        ),
+                        {"id": worker_article.article_id},
+                    )
+                elif change == "source":
+                    editor.execute(
+                        text("UPDATE sources SET approval_status='rejected' WHERE id=:id"),
+                        {"id": worker_article.source_id},
+                    )
+                elif change == "disabled_source":
+                    editor.execute(
+                        text("UPDATE sources SET enabled=false WHERE id=:id"),
+                        {"id": worker_article.source_id},
+                    )
+                else:
+                    editor.execute(
+                        text("UPDATE article_analysis_jobs SET lease_token=:token WHERE id=:id"),
+                        {"token": uuid4(), "id": worker_article.job_id},
+                    )
+        return prepared
+
+    monkeypatch.setattr(analysis_tasks, "_prepare_application", racing_prepare)
+    analysis_tasks._analyze(worker_article.job_id)
+    with database() as session:
+        article = session.get(Article, worker_article.article_id)
+        job = session.get(ArticleAnalysisJob, worker_article.job_id)
+        if change == "tag":
+            assert article.publication_status == "published" and job.outcome == "applied"
+        elif change == "disabled_source":
+            assert article.publication_status == "unpublished" and job.outcome == "applied"
+            assert "source_policy_manual" in job.result["publication_policy"]["reasons"]
+        else:
+            assert (
+                article.publication_status == "unpublished"
+                and not article.classification_provenance
+            )
+            assert job.outcome == (
+                "unapproved" if change == "source" else None if change == "lease" else "superseded"
+            )
+        if change in {"topic", "content"}:
+            jobs = session.scalars(select(ArticleAnalysisJob)).all()
+            assert len(jobs) == 2
+            replacement = next(item for item in jobs if item.id != job.id)
+            assert replacement.status == "queued"
+            if change == "content":
+                assert replacement.input_snapshot["text"] == article.summary
+            else:
+                assert replacement.catalog_hash != job.catalog_hash
+    assert len(worker_article.inferences) == 1
+
+
+def test_continuously_changing_catalog_uses_bounded_dependency_retry(
+    database, worker_article, unlocked_ranking, monkeypatch
+):
+    prepare = analysis_tasks._prepare_application
+
+    def racing_prepare(*args):
+        prepared = prepare(*args)
+        with database.begin() as editor:
+            editor.execute(text("UPDATE topics SET description=gen_random_uuid()::text"))
+        return prepared
+
+    monkeypatch.setattr(analysis_tasks, "_prepare_application", racing_prepare)
+    analysis_tasks._analyze(worker_article.job_id)
+    with database() as session:
+        article = session.get(Article, worker_article.article_id)
+        job = session.get(ArticleAnalysisJob, worker_article.job_id)
+        assert article.publication_status == "unpublished" and not article.classification_provenance
+        assert job.status == "queued" and job.error == "analysis_dependency_failure"
+        assert job.available_at > utcnow() and job.attempts == 1
+        assert not job.lease_token
+    assert unlocked_ranking.calls == 4  # One inference ranking plus three preparations.
+    assert len(worker_article.inferences) == 1
diff --git a/tests/test_analysis_tasks.py b/tests/test_analysis_tasks.py
index b961759ea..41f50f204 100644
--- a/tests/test_analysis_tasks.py
+++ b/tests/test_analysis_tasks.py
@@ -78,6 +78,7 @@ def begin(self):
         "tags": [],
     }
     monkeypatch.setattr(analysis_tasks, "catalog", lambda _: taxonomy)
+    monkeypatch.setattr(analysis_tasks, "snapshot_current", lambda *args: True)
     output = dict(
         outcome="ready",
         developer_relevance="relevant",