@@ -620,21 +620,27 @@ observed server revision; conflicting refreshes must
 rediscover before retrying. Imports carry the accepted revision, so an older snapshot cannot
 publish after a changed inventory. Only accepted publications establish missing Dags within
 a file; rejected requests and incomplete discovery do not establish absence. Every
-``[scheduler] parsing_cleanup_interval``, the scheduler marks the Dags of inactive or unassigned
-bundles stale, removes stale warnings, and ends expired processor Jobs, in bounded batches and in
+``[scheduler] parsing_cleanup_interval``, the scheduler marks the Dags of inactive bundles stale,
+removes stale warnings, and ends expired processor Jobs, in bounded batches and in
 both modes.
 
 Callback and priority requests remain stored while claimed. A lost claim response or
-acknowledgment can be retried, and requests from a retired Job become available to another
-processor. A callback acknowledgment confirms delivery, not successful execution of user
-code. As with other at-least-once delivery, a crash after a callback's side effect and before
-acknowledgment can repeat that side effect. A graceful shutdown acknowledges finished work
-before the processor's Job ends. When a callback's bundle version cannot be
-prepared, the processor retries with backoff for up to ``[dag_processor] job_heartbeat_timeout``
-and then acknowledges the callback as failed. Priority requests are assigned to an import
-started after their requested refresh. Only that import's publication or terminal failure
-acknowledges the request; an older in-flight import cannot consume it. Discovery after the
-requested refresh can also acknowledge a request if it confirms the definition is absent.
+acknowledgment can be retried. Claims use increasing sequences per Job and work type; the
+client waits for a claim response before sending the next sequence for that type.
+Retrying an empty claim remains empty, even if work arrives
+afterward, and retrying an older claim cannot consume newly requested work. Requests from a
+retired Job become available to another processor. A callback acknowledgment confirms
+delivery, not successful execution of user code. As with other at-least-once delivery, a crash
+after a callback's side effect and before acknowledgment can repeat that side effect. A graceful
+shutdown acknowledges finished work before the processor's Job ends. When a callback's bundle
+version cannot be prepared, the processor retries with backoff for up to
+``[dag_processor] job_heartbeat_timeout``
+and then acknowledges the request without running the callback. Priority requests are assigned
+to an import started after their requested refresh. Only that import's publication or terminal
+failure acknowledges the request; an older in-flight import cannot consume it. Discovery after
+the requested refresh can also acknowledge a request if it confirms the definition is absent.
+A new request for a file already being imported remains pending for a subsequent import,
+even if the earlier import has not yet been acknowledged.
 
 The manager retains a bounded queue of pending publications. Each loop it sends queued
 publications and acknowledgments for up to a second, stopping at the first failure that may
 
@@ -3519,12 +3519,12 @@ dag_processor:
         or listing a single bundle, so set this above the longest of those operations; the local
         health check must also tolerate them, through ``health_check_threshold`` and the probe's
         failure budget. It also bounds how long a processor retries preparing the bundle
-        version of a claimed callback before acknowledging the callback as failed.
+        version of a claimed callback before acknowledging it without running it.
+        A callback that ran before its acknowledgment was recorded can run again after the Job
+        times out, repeating its side effects. Callbacks should tolerate repeated execution.
         The scheduler reads this value to end Jobs and the processor reads it to stop, so set the
         same value for both; changing it only in the processor's configuration has no effect on
         when the scheduler ends the Job.
-        A callback that ran before its acknowledgment was recorded can run again after the Job
-        times out, repeating its side effects. Callbacks should tolerate repeated execution.
       version_added: 3.5.0
       type: integer
       example: ~
 
@@ -146,6 +146,7 @@ def __init__(
         self._completion_state: TerminalJobState | None = None
         self._completed = False
         self._pending_claims: dict[str, ProcessorWorkClaimBody] = {}
+        self._claim_sequences: dict[str, int] = {}
         super().__init__(base_url=base_url, token="", **kwargs)
 
     @property
@@ -411,6 +412,7 @@ def claim_work(
             kind,
             ProcessorWorkClaimBody(
                 claim_id=uuid7(),
+                claim_sequence=self._claim_sequences.get(kind, 0) + 1,
                 bundle_names=bundle_names,
                 limit=min(limit, 100),
             ),
@@ -424,13 +426,12 @@ def claim_work(
             timeout=self._get_bounded_timeout(),
         )
         work = [ProcessorWorkItem.model_validate(item) for item in response.json()]
+        self._claim_sequences[kind] = body.claim_sequence
         del self._pending_claims[kind]
         return work
 
-    def acknowledge_work(
-        self, kind: Literal["callbacks", "priority"], work: ProcessorWorkItem, *, failed: bool = False
-    ) -> None:
-        body = ProcessorWorkAckBody(claim_id=work.claim_id, state="failed" if failed else "success")
+    def acknowledge_work(self, kind: Literal["callbacks", "priority"], work: ProcessorWorkItem) -> None:
+        body = ProcessorWorkAckBody(claim_id=work.claim_id)
         self.request(
             "POST",
             f"jobs/{self._require_job_id()}/requested-work/{kind}/{work.id}/ack",
 
@@ -341,8 +341,8 @@ class DagFileProcessorManager(LoggingMixin):
     _deferred_api_callbacks: list[CallbackRequest] = attrs.field(factory=list, init=False)
     _deferred_callback_retries: dict[int, DeferredCallback] = attrs.field(factory=dict, init=False)
     _callback_attempts: dict[UUID, list[ProcessorWorkItem]] = attrs.field(factory=dict, init=False)
-    _priority_claims: dict[str, PriorityParseClaim] = attrs.field(factory=dict, init=False)
-    _pending_work_acks: list[tuple[Literal["callbacks", "priority"], ProcessorWorkItem, bool]] = attrs.field(
+    _priority_claims: dict[tuple[str, UUID], PriorityParseClaim] = attrs.field(factory=dict, init=False)
+    _pending_work_acks: list[tuple[Literal["callbacks", "priority"], ProcessorWorkItem]] = attrs.field(
         factory=list, init=False
     )
     _num_run: int = attrs.field(default=0, init=False)
@@ -810,7 +810,7 @@ def claim_priority_files(self) -> list[DagFileInfo]:
         available = (
             self._parallelism
             - len(self._priority_claims)
-            - sum(kind == "priority" for kind, _, _ in self._pending_work_acks)
+            - sum(kind == "priority" for kind, _ in self._pending_work_acks)
         )
         if available <= 0:
             return []
@@ -822,7 +822,7 @@ def claim_priority_files(self) -> list[DagFileInfo]:
             return []
         files = []
         for item in work:
-            self._priority_claims[item.id] = PriorityParseClaim(
+            self._priority_claims[item.id, item.claim_id] = PriorityParseClaim(
                 work=item, refresh_generation=self._bundle_refresh_generations[item.bundle_name] + 1
             )
             bundle = bundles[item.bundle_name]
@@ -897,7 +897,7 @@ def fetch_callbacks(self) -> list[CallbackRequest]:
             self.max_callbacks_per_loop
             - len(self._callback_claims)
             - sum(len(work) for work in self._callback_attempts.values())
-            - sum(kind == "callbacks" for kind, _, _ in self._pending_work_acks)
+            - sum(kind == "callbacks" for kind, _ in self._pending_work_acks)
         )
         if available <= 0:
             return list(self._deferred_api_callbacks)
@@ -922,9 +922,9 @@ def _acknowledge_requested_work(self, *, deadline: float | None = None) -> None:
         if deadline is None:
             deadline = time.monotonic() + _API_SEND_BUDGET_SECONDS
         while self._pending_work_acks and time.monotonic() < deadline:
-            kind, item, failed = self._pending_work_acks[0]
+            kind, item = self._pending_work_acks[0]
             try:
-                self.api_client.acknowledge_work(kind, item, failed=failed)
+                self.api_client.acknowledge_work(kind, item)
             except httpx2.HTTPStatusError as error:
                 if error.response.status_code >= 500:
                     self.log.warning("Unable to acknowledge requested work; retaining the claim")
@@ -944,13 +944,13 @@ def _acknowledge_requested_work(self, *, deadline: float | None = None) -> None:
 
     def _finish_callback_attempt(self, proc: BaseDagFileProcessorProcess) -> None:
         for item in self._callback_attempts.pop(proc.id, []):
-            self._pending_work_acks.append(("callbacks", item, proc._exit_code != 0))
+            self._pending_work_acks.append(("callbacks", item))
 
-    def _finish_priority_claims(self, attempt_id: UUID, *, failed: bool) -> None:
+    def _finish_priority_claims(self, attempt_id: UUID) -> None:
         """Acknowledge only requests assigned to this import, never newer requests for its file."""
         for key, claim in list(self._priority_claims.items()):
             if claim.attempt_id == attempt_id:
-                self._pending_work_acks.append(("priority", claim.work, failed))
+                self._pending_work_acks.append(("priority", claim.work))
                 del self._priority_claims[key]
 
     def _get_priority_claims(self, file: DagFileInfo) -> list[PriorityParseClaim]:
@@ -1065,7 +1065,7 @@ def _add_callback_to_queue(self, request: CallbackRequest) -> None:
         if get_claiming_importer(request.filepath, request.bundle_name) is not None:
             self._log_dropped_lang_sdk_callback(request)
             if item := self._callback_claims.pop(id(request), None):
-                self._pending_work_acks.append(("callbacks", item, True))
+                self._pending_work_acks.append(("callbacks", item))
                 self._remove_deferred_callback(request)
             return
         retry = self._deferred_callback_retries.get(id(request))
@@ -1098,7 +1098,7 @@ def _remove_deferred_callback(self, request: CallbackRequest) -> None:
         self._deferred_callback_retries.pop(id(request), None)
 
     def _defer_claimed_callback(self, request: CallbackRequest) -> None:
-        """Retry preparing a claimed callback's bundle with backoff, then acknowledge it as failed."""
+        """Retry preparing a claimed callback's bundle with backoff, then drop the unpreparable request."""
         now = time.monotonic()
         timeout = conf.getint("dag_processor", "job_heartbeat_timeout")
         retry = self._deferred_callback_retries.setdefault(id(request), DeferredCallback(now + timeout))
@@ -1111,7 +1111,7 @@ def _defer_claimed_callback(self, request: CallbackRequest) -> None:
                 request.bundle_version,
                 timeout,
             )
-            self._pending_work_acks.append(("callbacks", self._callback_claims.pop(id(request)), True))
+            self._pending_work_acks.append(("callbacks", self._callback_claims.pop(id(request))))
             self._remove_deferred_callback(request)
             return
         retry.attempts += 1
@@ -1214,7 +1214,7 @@ def _publish_bundle_inventory(self, bundle_name: str, known_files: dict[str, set
                 and item.relative_fileloc not in body.files
                 and not is_source_discovery_failed(item.relative_fileloc, failed_sources)
             ):
-                self._pending_work_acks.append(("priority", item, True))
+                self._pending_work_acks.append(("priority", item))
                 del self._priority_claims[key]
 
     def _discard_unusable_bundle_source(self, bundle_name: str) -> None:
@@ -1232,7 +1232,7 @@ def _discard_unusable_bundle_source(self, bundle_name: str) -> None:
         )
         for key, claim in list(self._priority_claims.items()):
             if claim.work.bundle_name == bundle_name and claim.attempt_id not in active_attempts:
-                self._pending_work_acks.append(("priority", claim.work, True))
+                self._pending_work_acks.append(("priority", claim.work))
                 del self._priority_claims[key]
 
     def _invalidate_bundle_parse_source(self, bundle_name: str) -> None:
@@ -1882,11 +1882,11 @@ def handle_parsing_result(
                     file.bundle_name,
                 )
                 self._record_failed_publication(file, next_stat)
-                self._finish_priority_claims(proc.id, failed=True)
+                self._finish_priority_claims(proc.id)
                 return
 
         if proc.parsing_result is None and not is_callback_only:
-            self._finish_priority_claims(proc.id, failed=True)
+            self._finish_priority_claims(proc.id)
         self._file_stats[file] = next_stat
 
     def _record_failed_publication(self, file: DagFileInfo, stat: DagFileStat) -> None:
@@ -1964,15 +1964,14 @@ def _publish_pending_results(self, *, deadline: float | None = None) -> None:
                     continue
                 self.log.exception("Failed to publish parsing result for %s", file)
                 self._record_failed_publication(file, pending.stat)
-                self._finish_priority_claims(pending.body.attempt_id, failed=True)
             else:
                 self._file_stats[file] = pending.stat
-                self._finish_priority_claims(pending.body.attempt_id, failed=False)
+            self._finish_priority_claims(pending.body.attempt_id)
             del self._pending_publications[file]
 
     def _build_rejection_report(self, pending: PendingDagPublication, reason: str) -> PendingDagPublication:
         """Replace a result that cannot be published with an import error users can see."""
-        self._finish_priority_claims(pending.body.attempt_id, failed=True)
+        self._finish_priority_claims(pending.body.attempt_id)
         self._dispatch_sequence += 1
         if len(reason) > _MAX_REJECTION_REASON_LENGTH:
             reason = reason[:_MAX_REJECTION_REASON_LENGTH] + "..."
@@ -2505,7 +2504,7 @@ def _kill_timed_out_processors(self):
                 )
                 self._file_stats[file] = stat
                 if not processor.had_callbacks:
-                    self._finish_priority_claims(processor.id, failed=True)
+                    self._finish_priority_claims(processor.id)
 
         # Clean up `self._processors` after iterating over it
         for proc in processors_to_remove:
 
@@ -30,6 +30,7 @@
 from airflow.api_fastapi.execution_api.datamodels.dag_parsing import (
     DagBundleInventoryBody,
     DagParseResultBody,
+    ProcessorWorkItem,
 )
 from airflow.api_fastapi.execution_api.datamodels.job import DagParseTokenBody, JobState, TerminalJobState
 from airflow.dag_processing import api_client
@@ -1079,3 +1080,59 @@ def test_inventory_waits_for_a_slow_write_but_connects_quickly(make_client):
     timeout = requests[-1].extensions["timeout"]
     assert timeout["connect"] <= 1
     assert timeout["read"] == client.timeout.read > 1
+
+
+@pytest.mark.parametrize("kind", ["callbacks", "priority"])
+@pytest.mark.parametrize(
+    "failure",
+    [
+        httpx2.ReadError("Lost claim response"),
+        httpx2.Response(503),
+        httpx2.Response(200, json=[{"id": "incomplete"}]),
+    ],
+    ids=["lost_response", "server_error", "invalid_response"],
+)
+def test_claim_sequence_survives_retries_and_renewal_then_advances_per_kind(make_client, kind, failure):
+    client, requests = make_client(
+        make_registration_response(),
+        failure,
+        make_registration_response(),
+        httpx2.Response(200, json=[]),
+        httpx2.Response(200, json=[]),
+        httpx2.Response(200, json=[]),
+    )
+    client.register_job()
+    with pytest.raises((httpx2.HTTPError, ValueError)):
+        client.claim_work(kind, ["bundle-a"], 2)
+
+    client.register_job()
+    assert client.claim_work(kind, ["bundle-b"], 5) == []
+    other_kind = "callbacks" if kind == "priority" else "priority"
+    assert client.claim_work(other_kind, ["bundle-a"], 2) == []
+    assert client.claim_work(kind, ["bundle-b"], 5) == []
+
+    assert requests[1].content == requests[3].content
+    original = json.loads(requests[1].content)
+    assert original["claim_sequence"] == 1
+    assert original["bundle_names"] == ["bundle-a"]
+    assert original["limit"] == 2
+    assert json.loads(requests[4].content)["claim_sequence"] == 1
+    subsequent = json.loads(requests[5].content)
+    assert subsequent["claim_sequence"] == 2
+    assert subsequent["claim_id"] != original["claim_id"]
+    assert subsequent["bundle_names"] == ["bundle-b"]
+    assert subsequent["limit"] == 5
+
+
+@pytest.mark.parametrize("kind", ["callbacks", "priority"])
+def test_work_acknowledgment_sends_only_the_claim_identity(make_client, kind):
+    client, requests = make_client(make_registration_response(), httpx2.Response(204))
+    client.register_job()
+    work = ProcessorWorkItem(
+        id=str(uuid7()), claim_id=uuid7(), bundle_name="bundle-a", relative_fileloc="dag.py"
+    )
+
+    client.acknowledge_work(kind, work)
+
+    assert requests[-1].url.path == f"/execution/jobs/1/requested-work/{kind}/{work.id}/ack"
+    assert json.loads(requests[-1].content) == {"claim_id": str(work.claim_id)}