From 825fc17cd8b6bfca00239318b080590975ff7f59 Mon Sep 17 00:00:00 2001
From: Jyotinder Singh <33001894+JyotinderSingh@users.noreply.github.com>
Date: Sun, 16 Aug 2026 00:54:47 +0530
Subject: [PATCH] Harden batch collections: cancellation, reattach, and error
 reporting

Four defects in kinetic.collections surfaced by a docs audit.

cancel() now stops the whole collection. It sets a flag the submission
loop reads, so inputs queued behind a bounded max_concurrent are dropped
instead of launched, and cancelled indices are marked not-retryable so
retries > 0 no longer resubmits a cancelled job when cancellation turns
its status NOT_FOUND. A job registered while cancel() was mid-snapshot is
cancelled by the submission loop, which closes that race.

attach_batch() decides completeness from the child indices the manifest
names, not from how many handles loaded. results(cleanup=True) deletes a
child's whole GCS prefix including handle.json, so a later reattach could
not rebuild those children and blocked wait()/results() forever behind a
poll thread with no timeout. A named child whose handle is gone is now
terminal and reported by the new BatchHandle.unavailable_children, and
poll_timeout defaults to 30 minutes. _all_accounted_for follows the same
rule: once submission is complete the jobs list is frozen, so
as_completed() no longer waits on slots that can never hold a job.

BatchError.failures holds only JobHandles, so the documented
"for job in e.failures: job.job_id" no longer raises AttributeError.
Inputs that fail at submission time are reported through the new
BatchError.submission_failures, and they raise BatchError rather than
being reachable only from the handle.

map(max_concurrent=None, retries=0, fail_fast=True) hands back a handle
immediately again. Once every input is launched, fail_fast alone leaves
the loop nothing to act on, so it no longer polls, and the
calling-thread decision mirrors that predicate.

Adds 23 tests across the four areas, each verified to fail against the
previous behaviour, and updates the batched-jobs guide and API reference.
---
 docs/api.rst                |   2 +-
 docs/guides/batched_jobs.md | 219 ++++++++++---
 kinetic/collections.py      | 395 +++++++++++++++++------
 kinetic/collections_test.py | 622 +++++++++++++++++++++++++++++++++++-
 4 files changed, 1087 insertions(+), 151 deletions(-)

diff --git a/docs/api.rst b/docs/api.rst
index fe078a46..6861617d 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -36,7 +36,7 @@ Batched Jobs
 .. autofunction:: kinetic.collections.map
 
 .. autoclass:: BatchHandle
-   :members: statuses, status_counts, wait, as_completed, results, failures, cancel, cleanup
+   :members: statuses, status_counts, wait, as_completed, results, failures, submission_failures, unavailable_children, cancel, cleanup
    :show-inheritance:
 
 .. autoclass:: BatchError
diff --git a/docs/guides/batched_jobs.md b/docs/guides/batched_jobs.md
index 1f808144..a0506c30 100644
--- a/docs/guides/batched_jobs.md
+++ b/docs/guides/batched_jobs.md
@@ -157,7 +157,9 @@ losses = batch.results(ordered=False)
   manifest is preserved so `attach_batch()` still works.
 - **`return_exceptions`** (`bool`, default `False`): When `True`, failed
   positions contain the exception object instead of raising
-  `BatchError`. When `False`, any failure raises `BatchError`.
+  `BatchError`. When `False`, any failure raises `BatchError`. A job
+  that fails and an input that fails at submission time both count as a
+  failure.
 
 :::{important}
 A `TimeoutError` does not cancel running jobs. They continue executing
@@ -191,26 +193,39 @@ results before the last inputs have been submitted.
 ## Handling failures
 
 When any job fails and `return_exceptions=False` (the default),
-`results()` raises a `BatchError`.
+`results()` raises a `BatchError`. An input that fails at submission
+time raises a `BatchError` too.
 
 ```python
 try:
   results = batch.results()
 except kinetic.BatchError as e:
-  print(
-    f"Batch {e.group_id}: {len(e.failures)} of {len(e.partial_results)} jobs failed"
-  )
+  print(e)  # Batch grp-a1b2c3d4: 2 of 8 jobs failed
   for job in e.failures:
-    print(f"  {job.job_id}: {job.status().value}")
+    print(f"  job {job.job_id}: {job.status().value}")
+  for index, exc in e.submission_failures.items():
+    print(f"  input {index} never started: {exc}")
   # e.partial_results has results at successful positions, None at failed ones
 ```
 
-`BatchError` provides three attributes:
+`BatchError` provides four attributes:
 
 - **`group_id`**: The batch identifier.
-- **`failures`**: List of `JobHandle` objects for the failed jobs.
-- **`partial_results`**: A list aligned with `inputs` where successful
-  positions contain the result and failed positions contain `None`.
+- **`failures`**: A list of `JobHandle` objects for the jobs that
+  started and then failed. The list holds only `JobHandle` objects, so
+  `job.job_id` and `job.status()` are always safe to call.
+- **`submission_failures`**: A dict that maps an input index to the
+  exception from the submission of that input. These inputs never
+  became jobs. They have no `JobHandle`, and they never appear in
+  `failures`.
+- **`partial_results`**: A list aligned with `inputs`. A successful
+  position holds the result. A failed position holds `None`.
+
+:::{note}
+`partial_results` aligns with `inputs` only for the default
+`ordered=True`. With `ordered=False`, it holds the results that
+`results()` collected, in completion order.
+:::
 
 ### Tolerating failures
 
@@ -228,17 +243,29 @@ for i, r in enumerate(results):
 
 ### Inspecting failures
 
-`failures()` returns handles for jobs with status `FAILED`. It
-intentionally excludes `NOT_FOUND` because that status is ambiguous —
-a job may be `NOT_FOUND` because its Kubernetes resources were cleaned
-up, not because it failed. Use `statuses()` for finer-grained
-inspection.
+`failures()` returns handles for the jobs with status `FAILED`. It
+excludes `NOT_FOUND`, because that status is ambiguous. A job can be
+`NOT_FOUND` because Kinetic cleaned up its Kubernetes resources, and
+not because the job failed. Use `statuses()` for a more exact view.
+
+After `results()` runs, `failures()` returns the failures from that
+collection pass, and not the live status of each job. This keeps the
+list correct after `cleanup=True` deletes the Kubernetes resources.
 
 ```python
 for job in batch.failures():
   print(f"{job.job_id}: {job.tail(n=20)}")
 ```
 
+`failures()` reports only the jobs that started. To see the inputs that
+failed before they became jobs, read `submission_failures`. It maps the
+input index to the exception from that submission.
+
+```python
+for index, exc in batch.submission_failures.items():
+  print(f"input {index} failed to submit: {exc}")
+```
+
 ## Retries
 
 The `retries` parameter specifies how many additional attempts a job
@@ -256,13 +283,16 @@ batch = train.run_async_map(configs, retries=2)
   Kubernetes resources (GCS artifacts are preserved for debugging).
 - The group manifest tracks the attempt count per job, so
   `attach_batch()` can distinguish retries from initial submissions.
-- Submission errors (when the call to the function itself raises) are
-  not retried. These are typically packaging or configuration errors
-  that would fail again.
+- Kinetic does not retry a submission error, which is an error that the
+  call to the function raises. These errors are usually packaging errors
+  or configuration errors, and they fail again.
+- Kinetic does not retry a cancelled job. `cancel()` marks its children,
+  so the `NOT_FOUND` status that cancellation causes never starts a new
+  attempt.
 
 :::{note}
-When `retries > 0`, job submission runs in a background thread so
-Kinetic can poll for failures and resubmit.
+When `retries > 0`, job submission runs in a background thread. This
+lets Kinetic poll for failures and submit the input again.
 :::
 
 ## Concurrency control
@@ -280,13 +310,19 @@ batch = train.run_async_map(configs, max_concurrent=8)
 batch = train.run_async_map(configs, max_concurrent=None)
 ```
 
-- **Default:** `64`. New jobs are launched as running ones finish.
-- **`None`:** All inputs are submitted immediately with no concurrency
-  limit. When combined with `retries=0` (the default), submission
-  happens synchronously in the calling thread before `map()` returns.
+- **Default:** `64`. Kinetic launches a new job each time a running job
+  finishes.
+- **`None`:** Kinetic submits all inputs immediately, with no
+  concurrency limit. The calling thread does this work when `retries=0`
+  and when `fail_fast` and `cancel_running_on_fail` are not both `True`.
+  See [Threading model](#threading-model).
 - Must be a positive integer when set. Passing `0` or a negative value
   raises `ValueError`.
 
+In every case `run_async_map()` returns the `BatchHandle` as soon as the
+submission work is handed off or complete. It never waits for the jobs
+to finish. Use `wait()` or `results()` when you want to block.
+
 :::{note}
 Kinetic logs a warning when submitting more than 100 jobs with
 `max_concurrent=None`, suggesting you set a limit to control resource
@@ -316,21 +352,36 @@ batch = train.run_async_map(
 )
 ```
 
-A "failure" here means either a submission error (
-the call raised) or a runtime failure (the remote job reached
-`FAILED` or `NOT_FOUND` status after exhausting retries).
+A failure is one of two events. The first is a submission error, when
+the call raises. The second is a runtime failure, when the remote job
+reaches `FAILED` or `NOT_FOUND` status after all of its attempts.
 
 ### Manual cancellation
 
-You can cancel all non-terminal jobs at any time, independent of the
-`fail_fast` setting:
+`cancel()` stops the full collection at any time. It is independent of
+the `fail_fast` setting.
 
 ```python
 batch.cancel()
 ```
 
-Cancellation deletes each job's Kubernetes resource but preserves GCS
-artifacts for debugging.
+`cancel()` does three things:
+
+- It deletes the Kubernetes resource of each job that is not terminal.
+  The GCS artifacts of that job stay in place for debugging.
+- It drops the inputs that `max_concurrent` holds in the queue. Kinetic
+  does not launch them.
+- It marks the children as cancelled. Kinetic does not submit them
+  again, even when `retries` is above zero.
+
+A cancelled job reports the status `NOT_FOUND`, because its Kubernetes
+resource is gone. `wait()` returns after each job that started is
+terminal, and the slot of an input that never launched stays `None`.
+
+A cancelled job has no result. `results()` therefore raises a
+`BatchError` that lists those jobs in `failures`. Use
+`results(return_exceptions=True)` to read the results of the children
+that finished before the cancellation.
 
 ## Reattaching to a batch
 
@@ -347,15 +398,19 @@ batch = kinetic.attach_batch("grp-a1b2c3d4")
 results = batch.results()
 ```
 
-`attach_batch()` downloads the group manifest from GCS and reconstructs
-a `JobHandle` for each child. Index alignment is preserved: if the
-original batch had 10 inputs and only 7 were submitted before a crash,
-the returned `batch.jobs` list has 10 entries with `None` in the 3
-unsubmitted slots.
+`attach_batch()` downloads the group manifest from GCS and rebuilds a
+`JobHandle` for each child. It keeps the index alignment. If the
+original batch had 10 inputs, and a crash stopped it after 7, the
+`batch.jobs` list still has 10 entries. The 3 empty slots hold `None`.
+
+If the manifest names fewer children than the batch expects, the
+original `map()` is still at work. The handle then polls the manifest in
+a background thread until the rest of the children appear, or until
+`poll_timeout` ends the poll.
 
 :::{note}
-Kinetic logs a warning when a reattached batch has fewer children than
-expected, indicating partial submission.
+Kinetic writes a warning when the manifest of a reattached batch names
+fewer children than expected. This shows a partial submission.
 :::
 
 **Parameters:**
@@ -365,6 +420,40 @@ expected, indicating partial submission.
   default when `None`.
 - **`cluster`** (`str | None`, default `None`): GKE cluster name. Uses
   the default when `None`.
+- **`poll_interval`** (`float`, default `10.0`): Seconds between
+  manifest polls while the batch is partially submitted.
+- **`poll_timeout`** (`float | None`, default `1800.0`): Maximum seconds
+  to poll for the remaining children. After the timeout, the handle
+  reports the submission as complete, and the empty slots stay `None`.
+  Reattach again to pick up the children that started since then.
+  `None` polls forever. Use `None` only when you are sure that the
+  original process is alive, because a dead submitter then blocks
+  `wait()` and `results()` forever.
+
+### Children that Kinetic cleaned up
+
+`results(cleanup=True)` deletes the GCS artifacts of each child that
+gives a result, and the `handle.json` file of the child is one of those
+artifacts. The group manifest stays in place, so `attach_batch()` still
+finds the batch. But it cannot rebuild a `JobHandle` for a child that it
+cleaned up.
+
+Kinetic treats such a child as terminal, and not as a child that is
+still on the way. The batch reports the submission as complete, and
+`wait()` and `results()` return immediately. The slot of that child
+stays `None`, and `results()` gives `None` at that position.
+
+`unavailable_children` shows which children are in this state. It maps
+the child index to the job ID from the manifest.
+
+```python
+batch = kinetic.attach_batch("grp-a1b2c3d4")
+print(batch.unavailable_children)
+# {0: 'job-1a2b3c4d', 1: 'job-5e6f7a8b'}
+```
+
+A `None` slot that `unavailable_children` does not name is an input that
+the original `map()` never submitted.
 
 ## Cleanup
 
@@ -381,6 +470,13 @@ manifest is preserved, so `attach_batch()` still works.
 results = batch.results()  # cleanup=True is the default
 ```
 
+:::{important}
+This cleanup deletes the result of each child. A later `attach_batch()`
+cannot collect those results a second time. Use `cleanup=False` when you
+want to reattach later and read the results again. See
+[Children that Kinetic cleaned up](#children-that-kinetic-cleaned-up).
+:::
+
 ### Full teardown
 
 To delete everything — all children's resources and the group manifest
@@ -406,15 +502,30 @@ via `attach_batch()` because the manifest has been deleted.
 
 ### Threading model
 
-When `max_concurrent` is set (the default is 64) or `retries > 0`,
-`run_async_map()` launches a non-daemon background thread to manage
-submissions. The thread polls active jobs for terminal states and
-launches new ones as concurrency slots free up. The `BatchHandle` is
-returned immediately.
+`run_async_map()` uses a non-daemon background thread when the
+submission loop must watch the jobs after it launches them. Three
+settings need this:
+
+- `max_concurrent` is set. The default is 64. The loop must wait for a
+  free slot before it launches the next input.
+- `retries` is above zero. The loop must see a failure before it can
+  submit that input again.
+- `fail_fast` and `cancel_running_on_fail` are both `True`. The loop
+  must see the first failure before it can cancel the siblings.
+
+In these cases the thread polls the active jobs, launches new jobs, and
+cancels jobs. `run_async_map()` returns the `BatchHandle` immediately.
+
+In all other cases the calling thread submits every input, and then
+`run_async_map()` returns. Kinetic starts no background thread, and the
+loop does not poll the jobs. A terminal status cannot change what the
+loop does next, so the loop stops as soon as the last input is
+submitted.
 
-When `max_concurrent=None` and `retries=0`, all jobs are submitted
-synchronously in the calling thread before `map()` returns. No
-background thread is created.
+`fail_fast` on its own is such a case. A submission error still stops
+the queue immediately, because the loop sees it inside the same
+submission pass. But after every input is launched, a runtime failure
+has nothing left for the loop to stop.
 
 ### Manifest
 
@@ -433,11 +544,17 @@ Each batch gets a unique identifier in the format `grp-{8-hex-chars}`
 
 ### Submission errors
 
-If a call to the function itself raises (e.g., a packaging or validation
-error), the exception is captured internally and the corresponding slot
-in `batch.jobs` remains `None`. These errors are surfaced when you call
-`results()` — either as entries in the `BatchError.partial_results`
-list or as exception objects when `return_exceptions=True`.
+A call to the function can raise, for example with a packaging error or
+a validation error. Kinetic then keeps the exception, and the related
+slot in `batch.jobs` stays `None`. Read these errors from
+`batch.submission_failures`, which maps the input index to the
+exception.
+
+`results()` reports them too. With `return_exceptions=True`, it puts the
+exception at that position in the result list. With
+`return_exceptions=False`, it raises a `BatchError` that holds the same
+map in `BatchError.submission_failures`. These inputs never became
+jobs, so `BatchError.failures` does not list them.
 
 ## Related pages
 
diff --git a/kinetic/collections.py b/kinetic/collections.py
index b604ce57..c2ccc4dc 100644
--- a/kinetic/collections.py
+++ b/kinetic/collections.py
@@ -32,6 +32,12 @@
 _STATUS_POLL_INTERVAL = 5.0
 _MANIFEST_POLL_INTERVAL = 10.0
 
+# Upper bound on how long `attach_batch()` waits for a still-submitting
+# `map()` to name its remaining children in the manifest.  A bound is
+# required: without one a stalled or dead submitter leaves `wait()` and
+# `results()` blocked forever.  Pass `poll_timeout=None` to opt out.
+_DEFAULT_MANIFEST_POLL_TIMEOUT = 1800.0
+
 
 def _resolve_bucket(
   project: str | None, cluster: str | None
@@ -50,9 +56,14 @@ class BatchError(Exception):
 
   Attributes:
     group_id: The collection's group identifier.
-    failures: List of JobHandles for failed children.
+    failures: List of JobHandles for children that were submitted and
+      then failed.  Never contains `None`, so `job.job_id` and
+      `job.status()` are always safe to call on its items.
     partial_results: List where successful positions contain the
       result and failed positions contain `None`.
+    submission_failures: Mapping of input index to the exception raised
+      while submitting that input.  Those inputs never became jobs, so
+      they have no `JobHandle` and never appear in `failures`.
   """
 
   def __init__(
@@ -60,11 +71,13 @@ def __init__(
     group_id: str,
     failures: list[JobHandle],
     partial_results: list[Any],
+    submission_failures: dict[int, Exception] | None = None,
   ):
     self.group_id = group_id
     self.failures = failures
     self.partial_results = partial_results
-    n_failed = len(failures)
+    self.submission_failures = dict(submission_failures or {})
+    n_failed = len(failures) + len(self.submission_failures)
     n_total = len(partial_results)
     super().__init__(f"Batch {group_id}: {n_failed} of {n_total} jobs failed")
 
@@ -109,6 +122,22 @@ class BatchHandle:
     default=None, repr=False, compare=False
   )
 
+  # Set by cancel().  The submission loop reads both: the event stops it
+  # launching queued inputs, and the index set stops it reading the
+  # resulting NOT_FOUND statuses as failures worth retrying.
+  _cancel_requested: threading.Event = field(
+    default_factory=threading.Event, repr=False, compare=False
+  )
+  _cancelled_indices: set[int] = field(
+    default_factory=set, repr=False, compare=False
+  )
+
+  # Children the manifest names but whose handle.json could not be
+  # downloaded (index -> job_id).  Populated by attach_batch().
+  _unavailable_children: dict[int, str] = field(
+    default_factory=dict, repr=False, compare=False
+  )
+
   def statuses(self) -> list[tuple[int, JobStatus]]:
     """Return `(index, status)` for each submitted job."""
     return [
@@ -120,15 +149,20 @@ def status_counts(self) -> dict[str, int]:
     return dict(collections.Counter(s.value for _, s in self.statuses()))
 
   def _all_accounted_for(self, seen: set[int]) -> bool:
-    """True when every job slot is either seen-terminal or a submission error."""
+    """True when no further job can reach a terminal state.
+
+    Submission being complete freezes the `jobs` list: a slot still
+    holding `None` will never hold a job.  Its input raised at
+    submission time, `cancel()` stopped it launching, or `attach_batch()`
+    could not load its handle.  Waiting on those slots would never end,
+    so what is left to wait for is every job that does exist reaching a
+    terminal state.
+    """
     if not self._submission_complete.is_set():
       return False
     with self._lock:
-      total_submitted = sum(1 for j in self.jobs if j is not None)
-    total_errors = len(self._submission_errors)
-    return len(seen) >= total_submitted and (
-      len(seen) + total_errors >= len(self.jobs)
-    )
+      submitted = {i for i, job in enumerate(self.jobs) if job is not None}
+    return submitted <= seen
 
   def wait(self, *, timeout: float | None = None) -> None:
     """Block until all jobs reach a terminal state."""
@@ -237,12 +271,25 @@ def results(
         as job statuses become `NOT_FOUND`.
       return_exceptions: If *True*, failed positions contain the
         exception object.  If *False*, raise `BatchError` on any
-        failure.
+        failure — including inputs that failed at submission time,
+        which are reported by `BatchError.submission_failures`.
 
     Returns:
       List of results (input order when *ordered=True*, completion
       order otherwise).
     """
+    unavailable = self.unavailable_children
+    if unavailable:
+      logging.warning(
+        "Batch %s: %d child(ren) at indices %s have no handle in GCS, so "
+        "their results cannot be collected; those positions are None. "
+        "Their artifacts were most likely already deleted by an earlier "
+        "results(cleanup=True) or cleanup() call.",
+        self.group_id,
+        len(unavailable),
+        sorted(unavailable),
+      )
+
     if ordered:
       results_list, failures = self._results_ordered(
         timeout=timeout, cleanup=cleanup, return_exceptions=return_exceptions
@@ -252,11 +299,13 @@ def results(
         timeout=timeout, cleanup=cleanup, return_exceptions=return_exceptions
       )
 
-    if failures and not return_exceptions:
+    submission_failures = self.submission_failures
+    if (failures or submission_failures) and not return_exceptions:
       raise BatchError(
         group_id=self.group_id,
         failures=failures,
         partial_results=results_list,
+        submission_failures=submission_failures,
       )
 
     return results_list
@@ -268,18 +317,20 @@ def _results_ordered(
     cleanup: bool,
     return_exceptions: bool,
   ) -> tuple[list[Any], list[JobHandle]]:
-    """Collect results in input order (waits for all jobs first)."""
+    """Collect results in input order (waits for all jobs first).
+
+    The returned failure list holds only real `JobHandle` objects.
+    Inputs that never became jobs are reported separately through
+    `submission_failures`.
+    """
     self.wait(timeout=timeout)
     failures: list[JobHandle] = []
     results_list: list[Any] = [None] * len(self.jobs)
 
     for i, job in enumerate(self.jobs):
       if job is None:
-        if i in self._submission_errors:
-          exc = self._submission_errors[i]
-          if return_exceptions:
-            results_list[i] = exc
-          failures.append(None)  # type: ignore[arg-type]
+        if return_exceptions and i in self._submission_errors:
+          results_list[i] = self._submission_errors[i]
         continue
       try:
         results_list[i] = job.result(cleanup=cleanup)
@@ -289,7 +340,7 @@ def _results_ordered(
         failures.append(job)
 
     with self._lock:
-      self._cached_failures = [f for f in failures if f is not None]
+      self._cached_failures = list(failures)
     return results_list, failures
 
   def _results_completion_order(
@@ -311,14 +362,12 @@ def _results_completion_order(
           results_list.append(exc)
         failures.append(job)
 
-    for idx in sorted(self._submission_errors):
-      exc = self._submission_errors[idx]
-      if return_exceptions:
-        results_list.append(exc)
-      failures.append(None)  # type: ignore[arg-type]
+    if return_exceptions:
+      for idx in sorted(self._submission_errors):
+        results_list.append(self._submission_errors[idx])
 
     with self._lock:
-      self._cached_failures = [f for f in failures if f is not None]
+      self._cached_failures = list(failures)
     return results_list, failures
 
   def failures(self) -> list[JobHandle]:
@@ -358,9 +407,44 @@ def submission_failures(self) -> dict[int, Exception]:
     with self._lock:
       return dict(self._submission_errors)
 
+  @property
+  def unavailable_children(self) -> dict[int, str]:
+    """Return an index-to-job-id map of children whose handle is missing.
+
+    Only `attach_batch()` fills this in.  Each entry is a child that the
+    group manifest names, and that was therefore submitted, whose
+    `handle.json` could not be downloaded.  Its GCS prefix is almost
+    always gone because an earlier `results(cleanup=True)` or
+    `cleanup()` deleted it, so its `jobs[idx]` slot stays `None` and its
+    result can never be collected again.
+
+    A `None` slot that this mapping does not name is a different case:
+    an input that the original `map()` never submitted.
+    """
+    with self._lock:
+      return dict(self._unavailable_children)
+
   def cancel(self) -> None:
-    """Cancel all non-terminal jobs in the collection."""
-    for job in self.jobs:
+    """Cancel the collection: stop launching, then stop what is running.
+
+    Cancellation covers the whole collection, not just the jobs that
+    happen to be live right now.  Queued inputs that a bounded
+    `max_concurrent` has not launched yet are dropped, and the cancelled
+    children are marked so that a batch running with `retries > 0` does
+    not read their `NOT_FOUND` status as a failure and resubmit them.
+
+    Already-terminal children are left alone.  Cancelling deletes each
+    child's Kubernetes resource and keeps its GCS artifacts.
+    """
+    self._cancel_requested.set()
+
+    with self._lock:
+      # Every slot is off-limits from here on: the submitted ones are
+      # cancelled below, and the rest must never launch.
+      self._cancelled_indices.update(range(len(self.jobs)))
+      snapshot = list(self.jobs)
+
+    for job in snapshot:
       if job is None:
         continue
       try:
@@ -402,39 +486,92 @@ def cleanup(self, *, k8s: bool = True, gcs: bool = True) -> None:
           )
 
 
+def _child_index(child: dict, total_expected: int) -> int | None:
+  """Return a manifest child's `group_index`, or `None` if unusable."""
+  idx = child.get("group_index")
+  if not isinstance(idx, int) or idx < 0 or idx >= total_expected:
+    return None
+  return idx
+
+
 def _load_child_handle(
   bucket_name: str,
   child: dict,
-  total_expected: int,
   project: str,
-) -> tuple[int, JobHandle] | None:
+) -> JobHandle | None:
   """Download and reconstruct a single child handle.
 
-  Returns `(group_index, handle)` on success, or `None` if the
-  child has an invalid index or the download fails.
+  Returns `None` when the handle cannot be read back, which for a child
+  the manifest already names means its GCS prefix is gone rather than
+  not written yet.
   """
-  idx = child["group_index"]
-  if not isinstance(idx, int) or idx < 0 or idx >= total_expected:
-    logging.warning(
-      "Invalid child index %r (total_expected=%d); skipping",
-      idx,
-      total_expected,
-    )
-    return None
   try:
     payload = storage.download_handle(
       bucket_name, child["job_id"], project=project
     )
-    return idx, JobHandle.from_dict(payload)
+    return JobHandle.from_dict(payload)
   except (google_exceptions.GoogleAPIError, KeyError, ValueError):
     logging.warning(
-      "Could not load handle for child job %s (index %d); skipping",
-      child["job_id"],
-      idx,
+      "Could not load handle for child job %s; skipping",
+      child.get("job_id"),
     )
     return None
 
 
+def _listed_indices(manifest: dict, total_expected: int) -> set[int]:
+  """Return the child indices the manifest already claims.
+
+  A claimed index is one the original `map()` has submitted, whether or
+  not its `handle.json` can still be downloaded.  This is what tells
+  "not submitted yet" (worth waiting for) apart from "submitted, handle
+  gone" (never going to load, e.g. after `results(cleanup=True)` deleted
+  the child's GCS prefix).
+  """
+  return {
+    idx
+    for child in manifest.get("children", [])
+    if (idx := _child_index(child, total_expected)) is not None
+  }
+
+
+def _hydrate_children(
+  handle: BatchHandle,
+  manifest: dict,
+  bucket_name: str,
+  project: str,
+  total_expected: int,
+) -> None:
+  """Fill *handle*'s empty slots from the manifest's children.
+
+  A named child whose `handle.json` cannot be downloaded is recorded in
+  `handle._unavailable_children` instead of being left indistinguishable
+  from one that was never submitted.  Slots that already hold a handle
+  are left alone, so this is safe to call repeatedly as the manifest
+  grows.
+  """
+  for child in manifest.get("children", []):
+    idx = _child_index(child, total_expected)
+    if idx is None:
+      logging.warning(
+        "Invalid child index %r (total_expected=%d); skipping",
+        child.get("group_index"),
+        total_expected,
+      )
+      continue
+
+    with handle._lock:
+      if handle.jobs[idx] is not None:
+        continue
+
+    job_handle = _load_child_handle(bucket_name, child, project)
+    with handle._lock:
+      if job_handle is None:
+        handle._unavailable_children[idx] = child.get("job_id", "<unknown>")
+      else:
+        handle.jobs[idx] = job_handle
+        handle._unavailable_children.pop(idx, None)
+
+
 def _manifest_poll_loop(
   handle: BatchHandle,
   bucket_name: str,
@@ -446,9 +583,13 @@ def _manifest_poll_loop(
 ) -> None:
   """Poll GCS manifest until all children appear, then set `_submission_complete`.
 
-  Used by `attach_batch()` when the manifest shows fewer children
-  than `total_expected`, indicating the original `map()` is still
-  submitting.
+  Used by `attach_batch()` when the manifest names fewer children than
+  `total_expected`, indicating the original `map()` is still submitting.
+
+  The loop stops once the manifest *names* every child, not once every
+  handle loads.  A named child whose handle cannot be downloaded is
+  already as resolved as it will ever get, so waiting on it would only
+  burn the timeout and leave `wait()` blocked in the meantime.
   """
   deadline = None if timeout is None else time.monotonic() + timeout
 
@@ -456,7 +597,8 @@ def _manifest_poll_loop(
     while True:
       if deadline is not None and time.monotonic() >= deadline:
         logging.warning(
-          "Timed out polling manifest for batch %s (%d/%d children)",
+          "Timed out polling manifest for batch %s (%d/%d children). "
+          "Reattach again to pick up any children submitted since.",
           group_id,
           sum(1 for j in handle.jobs if j is not None),
           total_expected,
@@ -473,21 +615,9 @@ def _manifest_poll_loop(
         logging.warning("Failed to poll manifest for batch %s", group_id)
         continue
 
-      for child in manifest.get("children", []):
-        idx = child.get("group_index")
-        if not isinstance(idx, int) or idx < 0 or idx >= total_expected:
-          continue
-        with handle._lock:
-          if handle.jobs[idx] is not None:
-            continue
-        result = _load_child_handle(bucket_name, child, total_expected, project)
-        if result is not None:
-          loaded_idx, job_handle = result
-          with handle._lock:
-            handle.jobs[loaded_idx] = job_handle
-
-      loaded = sum(1 for j in handle.jobs if j is not None)
-      if loaded >= total_expected:
+      _hydrate_children(handle, manifest, bucket_name, project, total_expected)
+
+      if len(_listed_indices(manifest, total_expected)) >= total_expected:
         break
   finally:
     handle._submission_complete.set()
@@ -537,22 +667,35 @@ def has_work(self) -> bool:
     """True while jobs remain to be submitted or are still running."""
     return bool(self.pending) or bool(self.active)
 
+  @property
+  def cancelled(self) -> bool:
+    """True once `BatchHandle.cancel()` has been called."""
+    return self.handle._cancel_requested.is_set()
+
   def can_submit_more(self) -> bool:
     """True when the next pending job is allowed to launch."""
-    if not self.pending or self.stop_launching:
+    if not self.pending or self.stop_launching or self.cancelled:
       return False
     return self.max_concurrent is None or len(self.active) < self.max_concurrent
 
   def needs_active_polling(self) -> bool:
     """True when the loop must poll active jobs itself.
 
-    When all jobs are submitted with no retries and `fail_fast`
-    is off, the caller uses `wait()`/`results()` to observe
-    terminal states, so the submission loop can exit early.
+    Polling only earns its keep when a terminal status would change what
+    the loop does next: launch a queued input, retry a failed attempt, or
+    cancel the running siblings.  When none of those apply the caller
+    observes terminal states through `wait()` / `results()` instead, so
+    the loop exits and stops holding up whoever called `map()`.
+
+    Note that `fail_fast` alone does not require polling: with nothing
+    left to launch and no siblings to cancel, a failure has no effect the
+    loop could act on.
     """
     if not self.active:
       return False
-    return bool(self.pending) or self.max_attempts > 1 or self.fail_fast
+    if self.pending or self.max_attempts > 1:
+      return True
+    return self.fail_fast and self.cancel_running_on_fail
 
   def trigger_fail_fast(self) -> None:
     """Stop launching new jobs and optionally cancel siblings."""
@@ -569,6 +712,7 @@ def _submit_available(state: _SubmissionState) -> None:
   `trigger_fail_fast` is called.
   """
   handle = state.handle
+  launched: list[int] = []
 
   while state.can_submit_more():
     idx = state.pending.popleft()
@@ -609,6 +753,7 @@ def _submit_available(state: _SubmissionState) -> None:
     with handle._lock:
       handle.jobs[idx] = job_handle
     state.active.add(idx)
+    launched.append(idx)
 
     append_child_to_manifest(
       state.manifest, idx, job_handle.job_id, state.attempt_counts[idx]
@@ -625,9 +770,15 @@ def _submit_available(state: _SubmissionState) -> None:
         "Failed to update manifest after submitting index %d", idx
       )
 
-  if state.stop_launching:
+  if state.stop_launching or state.cancelled:
     state.pending.clear()
 
+  if state.cancelled and launched:
+    # `cancel()` sets its flag before it snapshots `handle.jobs`, so any
+    # job registered after that snapshot is one `cancel()` could not see.
+    # Cancelling those here is what closes the window between the two.
+    _cancel_active(handle, set(launched))
+
 
 def _poll_and_handle_terminal(state: _SubmissionState) -> None:
   """Poll active jobs for terminal states; retry or trigger fail_fast."""
@@ -646,12 +797,22 @@ def _poll_and_handle_terminal(state: _SubmissionState) -> None:
     except (RuntimeError, google_exceptions.GoogleAPIError):
       logging.warning("Failed to poll status for index %d", idx)
 
+  with handle._lock:
+    cancelled_indices = set(handle._cancelled_indices)
+
   for idx, status, job in newly_terminal:
     state.active.discard(idx)
 
     if status not in (JobStatus.FAILED, JobStatus.NOT_FOUND):
       continue
 
+    if idx in cancelled_indices:
+      # Cancelling deletes the child's K8s resource, so its status turns
+      # NOT_FOUND.  That is the requested outcome — resubmitting it would
+      # undo the cancellation, and failing the batch over it would report
+      # a failure the caller asked for.
+      continue
+
     if state.attempt_counts[idx] < state.max_attempts:
       # Retry: clean up previous attempt's K8s resources and re-queue.
       try:
@@ -663,6 +824,27 @@ def _poll_and_handle_terminal(state: _SubmissionState) -> None:
       state.trigger_fail_fast()
 
 
+def _runs_in_calling_thread(
+  max_concurrent: int | None,
+  retries: int,
+  fail_fast: bool,
+  cancel_running_on_fail: bool,
+) -> bool:
+  """True when the submission loop can finish without outliving `map()`.
+
+  That needs every input to launch on the first pass (no concurrency
+  limit) *and* nothing that would keep the loop polling afterwards — no
+  retries to schedule, and no fail-fast cancellation to perform.  Any
+  other combination has to run in a background thread, or `map()` would
+  block until the whole batch finishes instead of returning a handle.
+
+  Mirrors `_SubmissionState.needs_active_polling`; keep the two in step.
+  """
+  if max_concurrent is not None or retries > 0:
+    return False
+  return not (fail_fast and cancel_running_on_fail)
+
+
 def _submission_loop(
   submit_fn,
   inputs: list,
@@ -677,8 +859,8 @@ def _submission_loop(
   """Core submission and retry loop.
 
   Mutates *handle.jobs* and *manifest* in place.  Runs in the calling
-  thread (`max_concurrent=None` and `retries=0`) or in a background
-  thread otherwise.
+  thread or in a background thread, as decided by
+  `_runs_in_calling_thread`.
 
   Each iteration follows three phases:
 
@@ -759,7 +941,10 @@ def map(
 
   Returns:
     A `BatchHandle` for observing, collecting, and cleaning up
-    the collection.
+    the collection.  Returns as soon as submission is either finished
+    in the calling thread or handed to a background thread — never
+    after the jobs themselves finish.  Use `wait()` or `results()` to
+    block on the batch.
   """
   if not callable(submit_fn):
     raise TypeError("submit_fn must be callable")
@@ -817,7 +1002,9 @@ def map(
       len(inputs),
     )
 
-  if max_concurrent is None and retries == 0:
+  if _runs_in_calling_thread(
+    max_concurrent, retries, fail_fast, cancel_running_on_fail
+  ):
     # Simple path: submit all in calling thread.
     _submission_loop(
       submit_fn=submit_fn,
@@ -857,17 +1044,25 @@ def attach_batch(
   project: str | None = None,
   cluster: str | None = None,
   poll_interval: float = _MANIFEST_POLL_INTERVAL,
-  poll_timeout: float | None = None,
+  poll_timeout: float | None = _DEFAULT_MANIFEST_POLL_TIMEOUT,
 ) -> BatchHandle:
   """Reattach to an existing batch collection by *group_id*.
 
   Downloads the group manifest from GCS, reconstructs `JobHandle`
   objects for each child, and returns a fully usable `BatchHandle`.
 
-  If the manifest has fewer children than `total_expected` (i.e.
-  the original `map()` is still submitting), the returned handle
-  polls the manifest in a background thread until all children
-  appear or *poll_timeout* is reached.
+  If the manifest names fewer children than `total_expected` (i.e. the
+  original `map()` is still submitting), the returned handle polls the
+  manifest in a background thread until the rest are named or
+  *poll_timeout* is reached.
+
+  A child the manifest names but whose `handle.json` cannot be
+  downloaded is *not* treated as still-pending — its GCS artifacts have
+  typically been cleaned up (by `results(cleanup=True)` or
+  `JobHandle.cleanup()`), so no amount of polling will produce it.  Its
+  slot stays `None` and the batch is reported as fully submitted, which
+  keeps `wait()` and `results()` from blocking on a job that no longer
+  exists.
 
   Args:
     group_id: The collection identifier (e.g. `"grp-a1b2c3d4"`).
@@ -877,8 +1072,11 @@ def attach_batch(
       active profile's cluster, then the built-in default.
     poll_interval: Seconds between manifest polls when the batch
       is partially submitted.
-    poll_timeout: Maximum seconds to poll for remaining children.
-      `None` means poll indefinitely.
+    poll_timeout: Maximum seconds to poll for remaining children,
+      30 minutes by default.  On timeout the handle reports submission
+      as complete and the missing slots stay `None`; reattach again to
+      pick up children submitted since.  `None` polls indefinitely and
+      risks blocking `wait()` forever if the submitter has died.
 
   Returns:
     A hydrated `BatchHandle` ready for `results()`, etc.
@@ -895,35 +1093,44 @@ def attach_batch(
   # Preallocate to total_expected and slot each child by group_index
   # so that index alignment is preserved even when some handles are
   # missing or the batch was only partially submitted.
-  jobs: list[JobHandle | None] = [None] * total_expected
-
-  for child in children:
-    result = _load_child_handle(
-      bucket_name, child, total_expected, resolved_project
-    )
-    if result is not None:
-      idx, job_handle = result
-      jobs[idx] = job_handle
-
   handle = BatchHandle(
     group_id=manifest["group_id"],
     name=manifest.get("group_name"),
     tags=manifest.get("tags", {}),
-    jobs=jobs,
+    jobs=[None] * total_expected,
     _bucket_name=bucket_name,
     _project=resolved_project,
   )
 
-  loaded = sum(1 for j in jobs if j is not None)
-  if loaded >= total_expected:
-    # All children present — mark complete immediately.
+  _hydrate_children(
+    handle, manifest, bucket_name, resolved_project, total_expected
+  )
+
+  unavailable = handle.unavailable_children
+  if unavailable:
+    logging.warning(
+      "Batch %s: %d of %d children have no handle in GCS (indices %s). "
+      "Their artifacts were already deleted — usually by an earlier "
+      "results(cleanup=True) — so their results cannot be collected "
+      "again and their slots stay None.",
+      group_id,
+      len(unavailable),
+      total_expected,
+      sorted(unavailable),
+    )
+
+  # Completeness is decided by what the manifest *names*, not by how
+  # many handles loaded.  A named child whose handle is gone will never
+  # load, so polling for it would only stall wait() and results().
+  listed = _listed_indices(manifest, total_expected)
+  if len(listed) >= total_expected:
     handle._submission_complete.set()
   else:
     logging.warning(
-      "Batch %s was partially submitted: %d of %d expected jobs. "
-      "Polling manifest for remaining children.",
+      "Batch %s was partially submitted: %d of %d expected jobs are "
+      "recorded in the manifest. Polling for the remaining children.",
       group_id,
-      loaded,
+      len(listed),
       total_expected,
     )
     thread = threading.Thread(
diff --git a/kinetic/collections_test.py b/kinetic/collections_test.py
index ab6c058f..0950c7e5 100644
--- a/kinetic/collections_test.py
+++ b/kinetic/collections_test.py
@@ -4,10 +4,13 @@
 import os
 import tempfile
 import threading
+import time
 from unittest import mock
 
-from absl.testing import absltest
+from absl.testing import absltest, parameterized
+from google.api_core import exceptions as google_exceptions
 
+from kinetic import collections as kinetic_collections
 from kinetic.collections import (
   BatchError,
   BatchHandle,
@@ -17,6 +20,21 @@
 from kinetic.job_status import JobStatus
 from kinetic.jobs import JobHandle
 
+# Captured before any test patches `time.sleep`, which the collections
+# module reaches through the shared `time` module object.
+_REAL_SLEEP = time.sleep
+
+
+def _tick(_seconds=0):
+  """Stand-in for `time.sleep` that yields instead of honouring the delay.
+
+  Poll loops that a test drives from another thread must keep turning
+  without burning a core, so this sleeps for a token millisecond rather
+  than returning instantly.
+  """
+  _REAL_SLEEP(0.001)
+
+
 # ------------------------------------------------------------------
 # Helpers
 # ------------------------------------------------------------------
@@ -910,8 +928,6 @@ def test_missing_child_handle_preserves_index(self):
     """Missing handle.json should leave a None at the correct index."""
     manifest = self._make_manifest(2)
 
-    from google.api_core import exceptions as google_exceptions
-
     def download_side_effect(bucket, job_id, project=None):
       if job_id == "job-1":
         raise google_exceptions.NotFound("gone")
@@ -1000,8 +1016,10 @@ def failing_submit(*args, **kwargs):
       with self.assertRaises(BatchError) as ctx:
         handle.results()
 
-    # Index 1 failed at submission time.
-    self.assertEqual(len(ctx.exception.failures), 1)
+    # Index 1 failed at submission time.  No job exists for it, so it is
+    # reported through submission_failures and never lands in failures.
+    self.assertEqual(ctx.exception.failures, [])
+    self.assertIsInstance(ctx.exception.submission_failures[1], RuntimeError)
 
   def test_submission_error_with_return_exceptions(self):
     """Per-input submission failures should appear as exceptions."""
@@ -1456,5 +1474,599 @@ def download_handle_side_effect(bucket, job_id, project=None):
     self.assertIsNotNone(handle.jobs[1])
 
 
+# ------------------------------------------------------------------
+# cancel() — stops launching, and cancelled jobs are never retried
+# ------------------------------------------------------------------
+
+
+class TestCancelStopsLaunching(absltest.TestCase):
+  """`cancel()` must cover queued inputs, not only running jobs."""
+
+  def _submit_fn(self, submitted, first_submitted):
+    def submit_fn(x):
+      submitted.append(x)
+      first_submitted.set()
+      return _make_handle(job_id=f"job-{x}")
+
+    submit_fn.__name__ = "fn"
+    return submit_fn
+
+  def test_cancel_drops_queued_inputs(self):
+    """With bounded concurrency, queued inputs must not launch after cancel."""
+    submitted = []
+    first_submitted = threading.Event()
+    cancelled = set()
+
+    def mock_status(self_handle):
+      if self_handle.job_id in cancelled:
+        return JobStatus.NOT_FOUND
+      return JobStatus.RUNNING
+
+    def track_cancel(self_handle):
+      cancelled.add(self_handle.job_id)
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+      mock.patch("kinetic.collections.time.sleep", _tick),
+      mock.patch.object(JobHandle, "status", mock_status),
+      mock.patch.object(JobHandle, "cancel", track_cancel),
+    ):
+      handle = map(
+        self._submit_fn(submitted, first_submitted),
+        [0, 1, 2],
+        max_concurrent=1,
+        project="proj",
+        cluster="cluster",
+      )
+      self.assertTrue(first_submitted.wait(timeout=5))
+      handle.cancel()
+      self.assertTrue(handle._submission_complete.wait(timeout=5))
+
+    # Only the job that was already running got submitted; indices 1 and
+    # 2 were still queued behind max_concurrent=1 and are dropped.
+    self.assertEqual(submitted, [0])
+    self.assertEqual(cancelled, {"job-0"})
+    self.assertIsNone(handle.jobs[1])
+    self.assertIsNone(handle.jobs[2])
+
+  def test_cancel_does_not_resubmit_with_retries(self):
+    """Cancelling turns a job NOT_FOUND; retries must not resurrect it."""
+    submitted = []
+    first_submitted = threading.Event()
+    cancelled = set()
+
+    def mock_status(self_handle):
+      if self_handle.job_id in cancelled:
+        return JobStatus.NOT_FOUND
+      return JobStatus.RUNNING
+
+    def track_cancel(self_handle):
+      cancelled.add(self_handle.job_id)
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+      mock.patch("kinetic.collections.time.sleep", _tick),
+      mock.patch.object(JobHandle, "status", mock_status),
+      mock.patch.object(JobHandle, "cancel", track_cancel),
+      mock.patch.object(JobHandle, "cleanup") as mock_cleanup,
+    ):
+      handle = map(
+        self._submit_fn(submitted, first_submitted),
+        [0],
+        max_concurrent=1,
+        retries=2,
+        project="proj",
+        cluster="cluster",
+      )
+      self.assertTrue(first_submitted.wait(timeout=5))
+      handle.cancel()
+      self.assertTrue(handle._submission_complete.wait(timeout=5))
+
+    self.assertEqual(submitted, [0])
+    # The retry path cleans up the previous attempt's K8s resources
+    # before it re-queues the index.  Nothing may go down that path, or
+    # the loop is still treating the cancelled job as a failed attempt.
+    mock_cleanup.assert_not_called()
+
+  def test_cancel_catches_job_launched_during_cancel(self):
+    """A slot filled after cancel() snapshotted it must still be cancelled.
+
+    `cancel()` sets its flag before reading `handle.jobs`, so a
+    submission already in flight lands after the snapshot.  Closing that
+    window is the submission loop's job.
+    """
+    entered_second = threading.Event()
+    release_second = threading.Event()
+
+    def submit_fn(x):
+      if x == 1:
+        entered_second.set()
+        self.assertTrue(release_second.wait(timeout=5))
+      return _make_handle(job_id=f"job-{x}")
+
+    submit_fn.__name__ = "fn"
+
+    cancelled = set()
+
+    def mock_status(self_handle):
+      if self_handle.job_id in cancelled:
+        return JobStatus.NOT_FOUND
+      return JobStatus.RUNNING
+
+    def track_cancel(self_handle):
+      cancelled.add(self_handle.job_id)
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+      mock.patch("kinetic.collections.time.sleep", _tick),
+      mock.patch.object(JobHandle, "status", mock_status),
+      mock.patch.object(JobHandle, "cancel", track_cancel),
+    ):
+      handle = map(
+        submit_fn,
+        [0, 1],
+        max_concurrent=2,
+        project="proj",
+        cluster="cluster",
+      )
+      self.assertTrue(entered_second.wait(timeout=5))
+      # job-1 has not been registered yet, so cancel() cannot see it.
+      handle.cancel()
+      release_second.set()
+      self.assertTrue(handle._submission_complete.wait(timeout=5))
+
+    self.assertEqual(cancelled, {"job-0", "job-1"})
+
+  def test_cancel_lets_as_completed_finish(self):
+    """Slots cancelled before launch must not stall as_completed()."""
+    handle = _make_batch_handle(n_jobs=1, submission_complete=False)
+    handle.jobs.extend([None, None])
+
+    cancelled = set()
+
+    def mock_status(self_handle):
+      if self_handle.job_id in cancelled:
+        return JobStatus.NOT_FOUND
+      return JobStatus.RUNNING
+
+    def track_cancel(self_handle):
+      cancelled.add(self_handle.job_id)
+
+    with (
+      mock.patch.object(JobHandle, "status", mock_status),
+      mock.patch.object(JobHandle, "cancel", track_cancel),
+      mock.patch("kinetic.collections.time.sleep"),
+    ):
+      handle.cancel()
+      # Stands in for the submission loop, which this test does not run.
+      handle._submission_complete.set()
+      yielded = list(handle.as_completed(poll_interval=0.01, timeout=5))
+
+    self.assertEqual(cancelled, {"job-0"})
+    # The two slots that never held a job must not stall the iterator.
+    self.assertEqual([j.job_id for j in yielded], ["job-0"])
+
+
+# ------------------------------------------------------------------
+# attach_batch() after the children were cleaned up
+# ------------------------------------------------------------------
+
+
+class TestAttachAfterChildCleanup(absltest.TestCase):
+  """`results(cleanup=True)` deletes each child's whole GCS prefix.
+
+  The group manifest survives, so a later `attach_batch()` sees children
+  it can name but whose `handle.json` is gone.  Those slots are terminal,
+  not pending — polling for them would block `wait()` forever.
+  """
+
+  def _manifest(self, n_children=2, total_expected=None):
+    if total_expected is None:
+      total_expected = n_children
+    return {
+      "group_id": "grp-abc12345",
+      "group_kind": "map",
+      "group_name": "test-batch",
+      "tags": {},
+      "created_at": "2026-03-28T10:00:00Z",
+      "total_expected": total_expected,
+      "submit_fn_name": "train",
+      "children": [
+        {"group_index": i, "job_id": f"job-{i}", "attempts": 1}
+        for i in range(n_children)
+      ],
+    }
+
+  def _attach(self, manifest, **kwargs):
+    with (
+      mock.patch(
+        "kinetic.collections.storage.download_manifest",
+        return_value=manifest,
+      ),
+      mock.patch(
+        "kinetic.collections.storage.download_handle",
+        side_effect=google_exceptions.NotFound("artifacts deleted"),
+      ),
+    ):
+      return attach_batch(
+        "grp-abc12345", project="proj", cluster="cluster", **kwargs
+      )
+
+  def test_completes_without_starting_a_poll_thread(self):
+    with mock.patch.object(threading.Thread, "start") as mock_start:
+      handle = self._attach(self._manifest(2))
+
+    mock_start.assert_not_called()
+    self.assertTrue(handle._submission_complete.is_set())
+    self.assertEqual(handle.jobs, [None, None])
+
+  def test_unavailable_children_names_the_missing_jobs(self):
+    handle = self._attach(self._manifest(2))
+    self.assertEqual(handle.unavailable_children, {0: "job-0", 1: "job-1"})
+    # The property hands back a copy, not the live mapping.
+    handle.unavailable_children[9] = "job-9"
+    self.assertNotIn(9, handle.unavailable_children)
+
+  def test_wait_returns_instead_of_blocking(self):
+    handle = self._attach(self._manifest(2))
+    with mock.patch("kinetic.collections.time.sleep") as mock_sleep:
+      handle.wait(timeout=5)
+    mock_sleep.assert_not_called()
+
+  def test_results_return_none_without_blocking(self):
+    handle = self._attach(self._manifest(2))
+    with mock.patch("kinetic.collections.time.sleep"):
+      results = handle.results(timeout=5)
+    self.assertEqual(results, [None, None])
+
+  def test_results_warns_about_unavailable_children(self):
+    handle = self._attach(self._manifest(2))
+    with (
+      mock.patch("kinetic.collections.time.sleep"),
+      mock.patch("kinetic.collections.logging") as mock_logging,
+    ):
+      handle.results(timeout=5)
+
+    warnings = [c[0][0] for c in mock_logging.warning.call_args_list]
+    self.assertTrue(
+      any("have no handle in GCS" in w for w in warnings),
+      f"expected an unavailable-children warning, got {warnings}",
+    )
+
+  def test_as_completed_terminates(self):
+    handle = self._attach(self._manifest(2))
+    with mock.patch("kinetic.collections.time.sleep"):
+      yielded = list(handle.as_completed(poll_interval=0.01, timeout=5))
+    self.assertEqual(yielded, [])
+
+  def test_genuinely_partial_batch_still_polls(self):
+    """A manifest short of total_expected is still worth waiting on."""
+    with mock.patch.object(threading.Thread, "start") as mock_start:
+      handle = self._attach(self._manifest(n_children=1, total_expected=3))
+
+    mock_start.assert_called_once()
+    self.assertFalse(handle._submission_complete.is_set())
+    # Cleared so the daemon thread that never started cannot strand
+    # anything waiting on this handle.
+    handle._submission_complete.set()
+
+  def test_poll_thread_gets_a_bounded_default_timeout(self):
+    recorded = {}
+
+    def fake_poll_loop(**kwargs):
+      recorded.update(kwargs)
+      kwargs["handle"]._submission_complete.set()
+
+    with mock.patch("kinetic.collections._manifest_poll_loop", fake_poll_loop):
+      handle = self._attach(self._manifest(n_children=1, total_expected=3))
+      self.assertTrue(handle._submission_complete.wait(timeout=5))
+
+    self.assertIsNotNone(recorded["timeout"])
+    self.assertEqual(
+      recorded["timeout"],
+      kinetic_collections._DEFAULT_MANIFEST_POLL_TIMEOUT,
+    )
+
+  def test_poll_loop_stops_once_every_child_is_named(self):
+    """Polling ends on manifest coverage, not on handles that load."""
+    partial = self._manifest(n_children=1, total_expected=2)
+    full = self._manifest(n_children=2, total_expected=2)
+
+    manifest_calls = [0]
+
+    def download_manifest_side_effect(bucket, group_id, project=None):
+      manifest_calls[0] += 1
+      return partial if manifest_calls[0] <= 1 else full
+
+    def download_handle_side_effect(bucket, job_id, project=None):
+      if job_id == "job-1":
+        raise google_exceptions.NotFound("artifacts deleted")
+      return {
+        "job_id": job_id,
+        "backend": "gke",
+        "project": "proj",
+        "cluster_name": "cluster",
+        "zone": "us-central1-a",
+        "namespace": "default",
+        "bucket_name": "proj-kn-cluster-jobs",
+        "k8s_name": f"kinetic-{job_id}",
+        "image_uri": "image:tag",
+        "accelerator": "cpu",
+        "func_name": "train",
+        "display_name": f"kinetic-train-{job_id}",
+        "created_at": "2026-03-28T10:00:00Z",
+        "group_id": "grp-abc12345",
+        "group_kind": "map",
+        "group_index": 0,
+      }
+
+    with (
+      mock.patch(
+        "kinetic.collections.storage.download_manifest",
+        side_effect=download_manifest_side_effect,
+      ),
+      mock.patch(
+        "kinetic.collections.storage.download_handle",
+        side_effect=download_handle_side_effect,
+      ),
+      mock.patch("kinetic.collections.time.sleep", _tick),
+    ):
+      handle = attach_batch(
+        "grp-abc12345",
+        project="proj",
+        cluster="cluster",
+        poll_interval=0.01,
+      )
+      self.assertTrue(handle._submission_complete.wait(timeout=5))
+
+    self.assertIsNotNone(handle.jobs[0])
+    self.assertIsNone(handle.jobs[1])
+    self.assertEqual(handle.unavailable_children, {1: "job-1"})
+
+
+# ------------------------------------------------------------------
+# BatchError — failures hold real handles, submission errors are separate
+# ------------------------------------------------------------------
+
+
+class TestBatchErrorFailureReporting(absltest.TestCase):
+  def _mixed_batch_handle(self):
+    """A batch with one submission error and one runtime failure."""
+
+    def failing_submit(x):
+      if x == 1:
+        raise RuntimeError("bad input")
+      return _make_handle(job_id=f"job-{x}")
+
+    failing_submit.__name__ = "fn"
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+    ):
+      return map(
+        failing_submit,
+        [0, 1, 2],
+        max_concurrent=None,
+        project="proj",
+        cluster="cluster",
+      )
+
+  def _collect(self, handle, **kwargs):
+    def mock_result(self_handle, cleanup=True):
+      if self_handle.job_id == "job-2":
+        raise ValueError("job-2 failed at runtime")
+      return 42
+
+    with (
+      mock.patch.object(JobHandle, "status", return_value=JobStatus.SUCCEEDED),
+      mock.patch.object(JobHandle, "result", mock_result),
+      mock.patch("kinetic.collections.time.sleep"),
+    ):
+      return handle.results(**kwargs)
+
+  def test_documented_failure_loop_does_not_raise(self):
+    """`for job in e.failures: job.job_id` is documented — it must work."""
+    handle = self._mixed_batch_handle()
+    with self.assertRaises(BatchError) as ctx:
+      self._collect(handle, cleanup=False)
+
+    job_ids = [job.job_id for job in ctx.exception.failures]
+    self.assertEqual(job_ids, ["job-2"])
+
+  def test_submission_errors_reach_the_caller(self):
+    handle = self._mixed_batch_handle()
+    with self.assertRaises(BatchError) as ctx:
+      self._collect(handle, cleanup=False)
+
+    err = ctx.exception
+    self.assertEqual(sorted(err.submission_failures), [1])
+    self.assertIsInstance(err.submission_failures[1], RuntimeError)
+    # Both kinds of failure are counted in the message.
+    self.assertIn("2 of 3", str(err))
+
+  def test_submission_error_alone_still_raises(self):
+    """A batch whose jobs all succeed but whose input failed must raise."""
+
+    def failing_submit(x):
+      if x == 1:
+        raise RuntimeError("bad input")
+      return _make_handle(job_id=f"job-{x}")
+
+    failing_submit.__name__ = "fn"
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+    ):
+      handle = map(
+        failing_submit,
+        [0, 1],
+        max_concurrent=None,
+        project="proj",
+        cluster="cluster",
+      )
+
+    with (
+      mock.patch.object(JobHandle, "status", return_value=JobStatus.SUCCEEDED),
+      mock.patch.object(JobHandle, "result", return_value=42),
+      mock.patch("kinetic.collections.time.sleep"),
+      self.assertRaises(BatchError) as ctx,
+    ):
+      handle.results(cleanup=False)
+
+    self.assertEqual(ctx.exception.failures, [])
+    self.assertIsInstance(ctx.exception.submission_failures[1], RuntimeError)
+
+  def test_completion_order_failures_are_handles_too(self):
+    handle = self._mixed_batch_handle()
+    with self.assertRaises(BatchError) as ctx:
+      self._collect(handle, ordered=False, cleanup=False)
+
+    # Named explicitly: an empty list would satisfy an all() check while
+    # hiding the completion-order path dropping failures altogether.
+    self.assertEqual([job.job_id for job in ctx.exception.failures], ["job-2"])
+    self.assertIsInstance(ctx.exception.submission_failures[1], RuntimeError)
+
+  def test_handle_exposes_the_same_submission_failures(self):
+    handle = self._mixed_batch_handle()
+    self.assertIsInstance(handle.submission_failures[1], RuntimeError)
+
+  def test_submission_failures_default_to_empty(self):
+    """The three-argument constructor stays valid."""
+    err = BatchError("grp-123", [_make_handle()], [None, 42])
+    self.assertEqual(err.submission_failures, {})
+
+  def test_submission_failures_are_copied(self):
+    source = {1: RuntimeError("bad input")}
+    err = BatchError("grp-123", [], [None, None], source)
+    source[2] = ValueError("injected")
+    self.assertEqual(sorted(err.submission_failures), [1])
+
+
+# ------------------------------------------------------------------
+# Threading decision — map() must hand back a handle, not block
+# ------------------------------------------------------------------
+
+
+class TestRunsInCallingThread(parameterized.TestCase):
+  @parameterized.named_parameters(
+    # (max_concurrent, retries, fail_fast, cancel_running_on_fail, expected)
+    ("plain", None, 0, False, False, True),
+    ("fail_fast_only", None, 0, True, False, True),
+    ("cancel_flag_without_fail_fast", None, 0, False, True, True),
+    ("fail_fast_and_cancel", None, 0, True, True, False),
+    ("bounded_concurrency", 8, 0, False, False, False),
+    ("retries", None, 1, False, False, False),
+    ("bounded_with_fail_fast", 8, 0, True, False, False),
+  )
+  def test_decision(
+    self, max_concurrent, retries, fail_fast, cancel_running_on_fail, expected
+  ):
+    self.assertEqual(
+      kinetic_collections._runs_in_calling_thread(
+        max_concurrent, retries, fail_fast, cancel_running_on_fail
+      ),
+      expected,
+    )
+
+  def test_fail_fast_returns_a_handle_without_waiting(self):
+    """fail_fast alone must not make map() wait for the jobs to finish.
+
+    With everything submitted, no retries, and no siblings to cancel, a
+    terminal status changes nothing the loop could act on — so the loop
+    has no reason to poll, and map() has no reason to block.
+    """
+    polled = []
+
+    def submit_fn(x):
+      return _make_handle(job_id=f"job-{x}")
+
+    submit_fn.__name__ = "fn"
+
+    def mock_status(self_handle):
+      polled.append(self_handle.job_id)
+      return JobStatus.SUCCEEDED
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+      mock.patch("kinetic.collections.time.sleep") as mock_sleep,
+      mock.patch.object(JobHandle, "status", mock_status),
+      mock.patch.object(threading.Thread, "start") as mock_start,
+    ):
+      handle = map(
+        submit_fn,
+        [0, 1, 2],
+        max_concurrent=None,
+        retries=0,
+        fail_fast=True,
+        project="proj",
+        cluster="cluster",
+      )
+
+    self.assertTrue(handle._submission_complete.is_set())
+    self.assertEqual(len([j for j in handle.jobs if j is not None]), 3)
+    # No background thread, and no waiting on the jobs it just launched.
+    mock_start.assert_not_called()
+    mock_sleep.assert_not_called()
+    self.assertEqual(polled, [])
+
+  def test_fail_fast_with_cancel_keeps_the_background_thread(self):
+    """Cancelling siblings needs someone watching for the first failure."""
+
+    def submit_fn(x):
+      return _make_handle(job_id=f"job-{x}")
+
+    submit_fn.__name__ = "fn"
+
+    cancelled = set()
+
+    def mock_status(self_handle):
+      if self_handle.job_id == "job-0":
+        return JobStatus.FAILED
+      if self_handle.job_id in cancelled:
+        return JobStatus.NOT_FOUND
+      return JobStatus.RUNNING
+
+    def track_cancel(self_handle):
+      cancelled.add(self_handle.job_id)
+
+    started_threads = []
+    original_start = threading.Thread.start
+
+    def capture_start(self_thread):
+      started_threads.append(self_thread)
+      original_start(self_thread)
+
+    with (
+      mock.patch("kinetic.collections.storage.upload_manifest"),
+      mock.patch("kinetic.collections.storage.upload_handle"),
+      mock.patch("kinetic.collections.time.sleep", _tick),
+      mock.patch.object(JobHandle, "status", mock_status),
+      mock.patch.object(JobHandle, "cancel", track_cancel),
+      mock.patch.object(threading.Thread, "start", capture_start),
+    ):
+      handle = map(
+        submit_fn,
+        [0, 1],
+        max_concurrent=None,
+        retries=0,
+        fail_fast=True,
+        cancel_running_on_fail=True,
+        project="proj",
+        cluster="cluster",
+      )
+      self.assertTrue(handle._submission_complete.wait(timeout=5))
+
+    self.assertEqual(cancelled, {"job-1"})
+    # The cancellation above can only happen off the calling thread.
+    self.assertEqual(len(started_threads), 1)
+    self.assertFalse(started_threads[0].daemon)
+
+
 if __name__ == "__main__":
   absltest.main()