fix(e2e): wait for managed batch cancellation before deleting inputs

tests/e2e/batches/COVERAGE.md

@@ -129,8 +129,9 @@ provider when deleted. Model-encoded and managed file IDs route themselves
 File deletion and batch cancellation check their responses and retry transient
 failures up to three times. Teardown attempts every registered cleanup before
 reporting failures as test errors. Already deleted files and batches that are
-terminal are safe to clean up again. Cancellation polls for up to ten minutes
-before input deletion, because accepting cancellation does not finish it
+terminal are safe to clean up again. Managed batch cancellation polls for up to eleven minutes
+before input deletion: the ten-minute provider window plus a propagation margin.
+Raw and model-encoded inputs can be deleted after cancellation is accepted
 
 Azure input uploads request `expires_after` anchored to `created_at` with
 `seconds=1209600`, and the lifecycle tests check the returned expiry. This is a

tests/e2e/batches/batch_cleanup.py

@@ -5,11 +5,12 @@ from typing import Final, Protocol
 from pydantic import BaseModel
 
 from batch_client import BatchObject, FileDeleteResponse
+from capabilities import is_managed_id
 from e2e_http import NetworkError, RateLimitedError, Result, Success, UnknownApiError
 
 CLEANUP_DELAYS: Final = (1.0, 2.0, 4.0)
 BATCH_TERMINAL_STATUSES: Final = frozenset({"completed", "failed", "expired", "cancelled"})
-BATCH_CANCEL_TIMEOUT_SECONDS: Final = 600.0
+BATCH_CANCEL_TIMEOUT_SECONDS: Final = 660.0
 BATCH_CANCEL_POLL_SECONDS: Final = 10.0
 
 
@@ -62,12 +63,15 @@ def cleanup_batch(
     wait: Callable[[float], None] = sleep,
     clock: Callable[[], float] = monotonic,
 ) -> None:
+    needs_terminal_state: Final = is_managed_id(batch_id)
     fetched: Final = _require_cleanup_success(
         cleanup_result(lambda: client.retrieve_batch(batch_id, key=key, provider=provider)),
         f"Retrieve batch {batch_id} for cleanup",
     )
     if fetched.status in BATCH_TERMINAL_STATUSES:
         return
+    if fetched.status == "cancelling" and not needs_terminal_state:
+        return
     if fetched.status != "cancelling":
         result: Final = cleanup_result(lambda: client.cancel_batch(batch_id, key=key, provider=provider))
         if not (isinstance(result, UnknownApiError) and result.status_code in {400, 409}):
@@ -75,6 +79,8 @@ def cleanup_batch(
             assert cancelled.status in BATCH_TERMINAL_STATUSES | {"cancelling"}, (
                 f"Cancel batch {batch_id} left status {cancelled.status}"
             )
+            if not needs_terminal_state:
+                return
     deadline: Final = clock() + BATCH_CANCEL_TIMEOUT_SECONDS
     while True:
         current = _require_cleanup_success(
@@ -84,6 +90,8 @@ def cleanup_batch(
         if current.status in BATCH_TERMINAL_STATUSES:
             return
         assert current.status == "cancelling", f"Cancel batch {batch_id} left status {current.status}"
+        if not needs_terminal_state:
+            return
         assert clock() < deadline, (
             f"Batch {batch_id} cancellation did not finish within {BATCH_CANCEL_TIMEOUT_SECONDS}s"
         )

tests/e2e/batches/test_batch_cleanup.py

@@ -12,6 +12,8 @@ from e2e_http import NetworkError, RateLimitedError, Result, Success, UnknownApi
 from lifecycle import ResourceManager
 from models import KeyGenerateBody
 
+MANAGED_BATCH_ID: Final = "bGl0ZWxsbV9wcm94eTtiYXRjaC0x"
+
 
 @dataclass
 class CleanupClient:
@@ -122,8 +124,8 @@ class TestBatchCancellation:
     def test_cancelling_batch_is_polled_until_terminal_without_cancelling_again(self) -> None:
         client: Final = CleanupClient(batches=iter((batch("cancelling"), batch("cancelling"), batch("cancelled"))))
         delays: Final[list[float]] = []
-        cleanup_batch(client, "batch-1", key="test-key", wait=delays.append)
-        assert client.calls == ["retrieve None batch-1"] * 3
+        cleanup_batch(client, MANAGED_BATCH_ID, key="test-key", wait=delays.append)
+        assert client.calls == [f"retrieve None {MANAGED_BATCH_ID}"] * 3
         assert delays == [10.0]
 
     def test_cancellation_timeout_is_reported_but_file_and_key_cleanup_still_run(self) -> None:
@@ -134,13 +136,13 @@ class TestBatchCancellation:
         manager: Final = ResourceManager(client=client, strict_cleanup=True)
         key: Final = manager.key()
         manager.defer(lambda: cleanup_file(client, "file-1", key=key))
-        manager.defer(lambda: cleanup_batch(client, "batch-1", key=key, clock=lambda: next(ticks)))
+        manager.defer(lambda: cleanup_batch(client, MANAGED_BATCH_ID, key=key, clock=lambda: next(ticks)))
         with pytest.raises(ExceptionGroup) as caught:
             manager.teardown()
         assert "cancellation did not finish" in str(caught.value.exceptions[0])
         assert client.calls == [
-            "retrieve None batch-1",
-            "retrieve None batch-1",
+            f"retrieve None {MANAGED_BATCH_ID}",
+            f"retrieve None {MANAGED_BATCH_ID}",
             "delete None file-1",
             "delete key test-key",
         ]
@@ -156,7 +158,7 @@ class TestBatchCancellation:
             batches=iter((batch("in_progress"), batch("cancelled"))), cancellations=iter((batch("cancelling"),))
         )
         cleanup_batch(client, "batch-1", key="test-key", provider="azure")
-        assert client.calls == ["retrieve azure batch-1", "cancel azure batch-1", "retrieve azure batch-1"]
+        assert client.calls == ["retrieve azure batch-1", "cancel azure batch-1"]
 
     @pytest.mark.parametrize("status", ["completed", "in_progress"])
     def test_cancellation_conflict_is_accepted_only_when_batch_became_inactive(self, status: str) -> None: