fix(plugins): a batch that could not be read is not a delivered batch

Two review findings from @karthik-indla on this PR.

_drain returned (0, True) on any read failure, so a claim nothing was posted
from counted as fully delivered. flush() then carried on to the next claim as
though this one had arrived, and the single signal that says the run went badly
never fired. The two cases are now separated: undecodable content is still
quarantined and reported delivered, because there is nothing left to send and
the rest of the run should continue, while an OSError leaves the file exactly
where it is and reports undelivered. Quarantining there would discard events
over a transient filesystem error, and nothing ever re-globs .corrupt.

Retries had no time backoff. _release_claim backdated straight to
immediately-reclaimable, so two senders meeting one momentary failure could walk
a batch from attempt 0 to the limit within seconds and discard it, when a retry
a minute later would have delivered. Releases now carry a cooldown that grows
with the attempts already spent, clamped so the mtime never lands in the future
and reads as a live lease.

Four tests: an unreadable batch is neither delivered nor quarantined,
undecodable content still is quarantined so one torn file cannot block every
later claim, and attempts cannot be burned without waiting. The expiry test now
ages the file between flushes, which is the wall time a real retry waits.

Claude-Session: https://claude.ai/code/session_01C7tEmH86HAr7GoAAKCEHZb

integrations/agent-plugin-core/python/telemetry.py

@@ -105,6 +105,11 @@ MAX_CLAIM_ATTEMPTS = 3
 # Parked claims drained per run, after the live spool. Bounded so a long backlog
 # cannot turn one flush into an unbounded send loop.
 MAX_PARKED_PER_RUN = 3
+# Added to the wait before a released claim becomes reclaimable, per attempt
+# already spent. Releasing straight to "reclaimable now" let two senders burn the
+# whole budget within seconds of one another on a single momentary failure, and
+# discard a batch a retry a minute later would have delivered.
+RETRY_COOLDOWN_SECONDS = 60
 
 
 def is_enabled() -> bool:
@@ -556,7 +561,11 @@ def _release_claim(claim: Path, remaining: list[dict[str, Any]]) -> None:
     if not _rewrite_claim(claim, remaining):
         return
     try:
-        released = time.time() - CLAIM_STALE_SECONDS - 1
+        # Backdate past the stale threshold so the next flush can pick it up,
+        # minus a cooldown that grows with the attempts already spent. Clamped so
+        # the mtime never lands in the future, which would read as a live lease.
+        cooldown = min(_claim_attempt(claim) * RETRY_COOLDOWN_SECONDS, CLAIM_STALE_SECONDS)
+        released = time.time() - CLAIM_STALE_SECONDS - 1 + cooldown
         os.utime(claim, (released, released))
     except OSError:
         pass
@@ -642,11 +651,13 @@ def _drain(claim: Path | None) -> tuple[int, bool]:
         return 0, True
     try:
         lines = claim.read_text(encoding="utf-8").splitlines()
-    except (OSError, ValueError):
-        # ValueError covers UnicodeDecodeError from a torn write. Quarantine
-        # rather than retry: flush() runs from a bare `finally:` in
+    except ValueError:
+        # UnicodeDecodeError from a torn write: the content is unrecoverable, so
+        # quarantine rather than retry. flush() runs from a bare `finally:` in
         # flush_worker, so raising here also skips the handoff cleanup, and an
         # undecodable file would otherwise be re-read on every flush forever.
+        # Reported as delivered because there is nothing left to deliver and the
+        # rest of the run should continue.
         try:
             claim.replace(claim.with_suffix(".corrupt"))
         except OSError:
@@ -655,6 +666,13 @@ def _drain(claim: Path | None) -> tuple[int, bool]:
             except OSError:
                 pass
         return 0, True
+    except OSError:
+        # Could not read it, which is not the same as having nothing to send.
+        # The file is left exactly where it is: a vanished or briefly unreadable
+        # claim is retryable, and quarantining it here would discard events over
+        # a transient filesystem error. Reported as undelivered so the run stops
+        # instead of counting a batch nothing was posted from as delivered.
+        return 0, False
     events = []
     for line in lines:
         try: