diff --git a/dev/trainer_rank_planner_design.md b/dev/trainer_rank_planner_design.md
index 9be81bb35..533d12c26 100644
--- a/dev/trainer_rank_planner_design.md
+++ b/dev/trainer_rank_planner_design.md
@@ -909,5 +909,31 @@ whose recorded failure type is `OutOfMemoryError`). Beyond the smaller cap,
 planning reports are dropped with the existing warning/failure counter. Driver
 delivery retains its original limits; this is not a cross-process storage quota.
 
+An execution owner can explicitly assign `RetentionLimits` using
+`report_retention_scope`. This chooses a private spool and cumulative report/byte
+allowance before emission; no scope keeps standalone behavior. An explicit
+`capture=False` suppresses report creation before replay/source hashing or I/O,
+including during construction before a rank can be bound. It differs from `None`
+(standalone retention) and a zero grant (which still accounts for omitted attempts).
+Suppression survives nested retention scopes and restores on exit; it follows
+ordinary ContextVar propagation, including `asyncio.to_thread`, but not arbitrary
+new threads. It does not disable the planner or change admission/training.
+An atomic, fsynced
+charge ledger precedes each payload write. Exact duplicate bytes cost nothing new,
+but deletion, failed payload writes and process restart never replenish the grant.
+Changed execution, producer or allowance identity refuses. Exhaustion keeps bounded
+omitted-attempt/byte counters, including a zero allowance and ordinary failures
+following a charge, without another spool. These count failed persistence attempts,
+not unique lost reports or certain payload absence. Counter writes are best effort:
+an unreadable ledger, disk failure or process death can leave coverage incomplete.
+Unknown crash leftovers are retained and count against occupancy; a first ledger
+write interrupted before publication requires owner reconciliation, not automatic
+cleanup or a new grant. Enter the context in the actual reporting thread: plain
+threads and `run_in_executor` do not inherit it. Spools belong to one producer;
+the file lock can block if that contract is violated or storage stalls.
+Ledger/lock metadata needs a separate owner reservation. This hook neither reserves
+capacity across producers nor acknowledges transport: those are the execution
+owner's responsibilities. It adds no GPU operation or full-input capture.
+
 Caladan's format-2 reader must land before this producer is enabled, and both
 rank and driver need the updated ART validator. Old format-1 JSON remains readable.
diff --git a/src/art/trainer_rank/_planner_misses.py b/src/art/trainer_rank/_planner_misses.py
index 228a06d11..2018c6186 100644
--- a/src/art/trainer_rank/_planner_misses.py
+++ b/src/art/trainer_rank/_planner_misses.py
@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Mapping
+from contextlib import nullcontext
 from dataclasses import asdict, dataclass
 from datetime import datetime, timezone
 import hashlib
@@ -30,7 +31,9 @@
 from typing import Any
 import uuid
 
-from . import _planner_evidence
+from . import _planner_evidence, _planner_retention
+from ._planner_retention import RetentionLimits as RetentionLimits
+from ._planner_retention import report_retention_scope as report_retention_scope
 
 ALLOW_OVERSIZED_ENV = "ART_TRAINER_RANK_ALLOW_OVERSIZED_BATCHES"
 MISS_THRESHOLD_ENV = "ART_TRAINER_RANK_PLANNER_MISS_THRESHOLD_PCT"
@@ -48,6 +51,7 @@
     "_planner_misses.py",
     "_gdn_memory.py",
     "_planner_evidence.py",
+    "_planner_retention.py",
 )
 _SOURCE_BYTE_LIMIT = 1024 * 1024
 _REPORT_KEYS = frozenset(
@@ -286,24 +290,51 @@ def pairs(items: list[tuple[str, Any]]) -> dict[str, Any]:
 
 
 def persist_report(
-    raw: bytes, spool_dir: Path, *, planning_budget: bool = False
+    raw: bytes,
+    spool_dir: Path,
+    *,
+    planning_budget: bool = False,
+    retention: RetentionLimits | None = None,
 ) -> Path:
     """Durably retain exact bytes; duplicate delivery is safe, conflicts refuse."""
     record = validate_report(raw)
-    with _spool_lock:
+    if retention is not None and spool_dir != retention.spool_dir:
+        raise ValueError("planner report spool differs from assigned allowance")
+    charged = (
+        _planner_retention.charge(
+            retention,
+            record["id"],
+            raw,
+            count_limit=MAX_PLANNING_REPORTS if planning_budget else MAX_SPOOL_REPORTS,
+            byte_limit=MAX_PLANNING_SPOOL_BYTES if planning_budget else MAX_SPOOL_BYTES,
+        )
+        if retention is not None
+        else nullcontext()
+    )
+    with _spool_lock, charged:
         spool_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
         info = spool_dir.lstat()
         if not stat.S_ISDIR(info.st_mode) or info.st_mode & 0o077:
             raise ValueError("report spool must be a private directory")
         path = spool_dir / f"{record['id']}.json"
         if path.exists() or path.is_symlink():
-            info = path.lstat()
-            if (
-                not stat.S_ISREG(info.st_mode)
-                or info.st_size != len(raw)
-                or path.read_bytes() != raw
-            ):
-                raise ValueError("existing report identity has different bytes")
+            descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+            with os.fdopen(descriptor, "rb") as existing:
+                info = os.fstat(existing.fileno())
+                if (
+                    not stat.S_ISREG(info.st_mode)
+                    or info.st_size != len(raw)
+                    or existing.read(len(raw) + 1) != raw
+                ):
+                    raise ValueError("existing report identity has different bytes")
+                # A prior attempt may have linked the file but failed its
+                # durability barrier. Visibility alone cannot acknowledge it.
+                os.fsync(existing.fileno())
+            directory = os.open(spool_dir, os.O_RDONLY | os.O_DIRECTORY)
+            try:
+                os.fsync(directory)
+            finally:
+                os.close(directory)
             return path
         # Rank-side ordinary planning failures may use only the first small
         # part of the spool. OOMs/misses and delivery keep their original limit.
@@ -317,8 +348,17 @@ def persist_report(
             if planning_budget
             else MAX_SPOOL_BYTES
         )
+        if retention is not None:
+            count_limit = min(count_limit, retention.max_reports)
+            byte_limit = min(byte_limit, retention.max_bytes)
         size = count = 0
         for entry in spool_dir.iterdir():
+            if retention is not None and entry.name in {
+                ".retention.lock",
+                ".retention.json",
+            }:
+                # Assigned metadata has its separate reserved allowance.
+                continue
             try:
                 item = entry.lstat()
             except FileNotFoundError:
@@ -378,7 +418,7 @@ def report(
     ) -> Path | None:
         """Serialize only a miss; ordinary observation failures never escape."""
         threshold = self.threshold_pct
-        if threshold is None:
+        if threshold is None or not _planner_retention.capture_enabled():
             return None
         event = event or ("oom" if oom else "estimate_miss")
         planning = event in {"admission_refused", "planning_error"}
@@ -456,11 +496,13 @@ def report(
                     f"replay unavailable: {'ValueError' if isinstance(exc, _ReportTooLarge) else type(exc).__name__}"
                 ]
                 raw = _encode(record)
+            retention = _planner_retention.current_limits()
             path = persist_report(
                 raw,
-                self.spool_dir,
+                self.spool_dir if retention is None else retention.spool_dir,
                 planning_budget=planning
                 and not (failure is not None and failure["type"] == "OutOfMemoryError"),
+                retention=retention,
             )
         except Exception as exc:
             self.failures += 1
diff --git a/src/art/trainer_rank/_planner_retention.py b/src/art/trainer_rank/_planner_retention.py
new file mode 100644
index 000000000..128cdfac8
--- /dev/null
+++ b/src/art/trainer_rank/_planner_retention.py
@@ -0,0 +1,208 @@
+"""Execution-assigned cumulative report allowances; no transport or quota refund."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager, suppress
+from contextvars import ContextVar
+from dataclasses import dataclass
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import tempfile
+from typing import Any, Iterator
+
+_UUID = re.compile(r"[0-9a-f]{32}\Z")
+_SHA256 = re.compile(r"[0-9a-f]{64}\Z")
+_LEDGER_LIMIT = 512 * 1024
+
+
+@dataclass(frozen=True)
+class RetentionLimits:
+    spool_dir: Path
+    max_bytes: int
+    max_reports: int
+    allowance_id: str
+    execution_id: str
+    producer_id: str
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.spool_dir, Path) or not self.spool_dir.is_absolute():
+            raise ValueError("assigned planner spool must be an absolute path")
+        if (
+            type(self.max_bytes) is not int
+            or not 0 <= self.max_bytes <= 256 * 1024 * 1024
+            or type(self.max_reports) is not int
+            or not 0 <= self.max_reports <= 1024
+            or any(
+                not isinstance(value, str) or not _UUID.fullmatch(value)
+                for value in (self.allowance_id, self.execution_id, self.producer_id)
+            )
+        ):
+            raise ValueError("invalid assigned planner allowance")
+
+    def identity(self) -> dict[str, Any]:
+        return {
+            "format": 1,
+            "allowance_id": self.allowance_id,
+            "execution_id": self.execution_id,
+            "producer_id": self.producer_id,
+            "max_bytes": self.max_bytes,
+            "max_reports": self.max_reports,
+        }
+
+
+_current: ContextVar[tuple[RetentionLimits | None, bool]] = ContextVar(
+    "planner_retention", default=(None, True)
+)
+
+
+@contextmanager
+def report_retention_scope(
+    limits: RetentionLimits | None, *, capture: bool = True
+) -> Iterator[None]:
+    if type(capture) is not bool:
+        raise ValueError("planner capture must be a boolean")
+    if limits is not None and not isinstance(limits, RetentionLimits):
+        raise ValueError("invalid assigned planner retention scope")
+    token = _current.set((limits, capture and capture_enabled()))
+    try:
+        yield
+    finally:
+        _current.reset(token)
+
+
+def current_limits() -> RetentionLimits | None:
+    return _current.get()[0]
+
+
+def capture_enabled() -> bool:
+    return _current.get()[1]
+
+
+def _encode(value: dict[str, Any]) -> bytes:
+    raw = (
+        json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n"
+    ).encode()
+    if len(raw) > _LEDGER_LIMIT:
+        raise ValueError("planner charge ledger exceeds limit")
+    return raw
+
+
+def _write(path: Path, value: dict[str, Any]) -> None:
+    raw = _encode(value)
+    descriptor, name = tempfile.mkstemp(prefix=".retention-", dir=path.parent)
+    temporary = Path(name)
+    try:
+        with os.fdopen(descriptor, "wb") as target:
+            target.write(raw)
+            target.flush()
+            os.fsync(target.fileno())
+        os.replace(temporary, path)
+        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+@contextmanager
+def charge(
+    limits: RetentionLimits,
+    event_id: str,
+    raw: bytes,
+    *,
+    count_limit: int,
+    byte_limit: int,
+) -> Iterator[None]:
+    """Commit a charge before the payload write; ambiguous writes never refund it.
+
+    Payload reclamation does not change this ledger. The caller reserves bounded
+    ledger/lock metadata separately from cumulative captured payload bytes.
+    """
+    if not _UUID.fullmatch(event_id):
+        raise ValueError("invalid planner charge identity")
+    root = limits.spool_dir
+    root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    info = root.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_mode & 0o077:
+        raise ValueError("assigned planner spool must be a private directory")
+    lock = os.open(
+        root / ".retention.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600
+    )
+    try:
+        if not stat.S_ISREG(os.fstat(lock).st_mode):
+            raise ValueError("planner retention lock is not regular")
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        path = root / ".retention.json"
+        try:
+            descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+        except FileNotFoundError:
+            if any(entry.name != ".retention.lock" for entry in root.iterdir()):
+                raise ValueError("assigned spool contains unaccounted evidence")
+            ledger = {
+                "allowance": limits.identity(),
+                "charges": {},
+                "omitted": 0,
+                "omitted_bytes": 0,
+            }
+        else:
+            with os.fdopen(descriptor, "rb") as source:
+                if not stat.S_ISREG(os.fstat(source.fileno()).st_mode):
+                    raise ValueError("planner charge ledger is not regular")
+                saved = source.read(_LEDGER_LIMIT + 1)
+            ledger = json.loads(saved)
+            if (
+                _encode(ledger) != saved
+                or set(ledger) != {"allowance", "charges", "omitted", "omitted_bytes"}
+                or _encode(ledger["allowance"]) != _encode(limits.identity())
+                or not isinstance(ledger["charges"], dict)
+                or type(ledger["omitted"]) is not int
+                or not 0 <= ledger["omitted"] <= 2**63 - 1
+                or type(ledger["omitted_bytes"]) is not int
+                or not 0 <= ledger["omitted_bytes"] <= 2**63 - 1
+            ):
+                raise ValueError("planner charge ledger differs from enrollment")
+        charges = ledger["charges"]
+        total = 0
+        for identifier, item in charges.items():
+            if (
+                not _UUID.fullmatch(identifier)
+                or not isinstance(item, list)
+                or len(item) != 2
+                or not isinstance(item[0], str)
+                or not _SHA256.fullmatch(item[0])
+                or type(item[1]) is not int
+                or not 0 < item[1] <= 16 * 1024 * 1024
+            ):
+                raise ValueError("invalid retained planner charge")
+            total += item[1]
+        if len(charges) > limits.max_reports or total > limits.max_bytes:
+            raise ValueError("planner charge ledger exceeds enrollment")
+        identity = [hashlib.sha256(raw).hexdigest(), len(raw)]
+        if event_id in charges and charges[event_id] != identity:
+            raise ValueError("planner report ID has conflicting charged bytes")
+        try:
+            if event_id not in charges:
+                if len(charges) >= min(limits.max_reports, count_limit) or total + len(
+                    raw
+                ) > min(limits.max_bytes, byte_limit):
+                    raise ValueError("assigned planner retention exhausted")
+                charges[event_id] = identity
+                _write(path, ledger)
+            yield
+        except Exception:
+            # Count failed attempts even after charging (including duplicate
+            # retries blocked by crash leftovers); never refund uncertain writes.
+            ledger["omitted"] = min(ledger["omitted"] + 1, 2**63 - 1)
+            ledger["omitted_bytes"] = min(ledger["omitted_bytes"] + len(raw), 2**63 - 1)
+            with suppress(Exception):
+                _write(path, ledger)
+            raise
+    finally:
+        os.close(lock)
diff --git a/tests/unit/test_planner_retention_budget.py b/tests/unit/test_planner_retention_budget.py
new file mode 100644
index 000000000..fdfe69c1d
--- /dev/null
+++ b/tests/unit/test_planner_retention_budget.py
@@ -0,0 +1,361 @@
+"""Assigned quotas count captured bytes once, including after source reclamation."""
+
+import asyncio
+from dataclasses import replace
+import json
+import stat
+
+import pytest
+
+from art.trainer_rank import _planner_misses as reports
+
+
+def limits(tmp_path, **values):
+    return reports.RetentionLimits(
+        spool_dir=tmp_path / "assigned",
+        max_bytes=values.pop("max_bytes", 1024 * 1024),
+        max_reports=values.pop("max_reports", 4),
+        allowance_id="a" * 32,
+        execution_id="b" * 32,
+        producer_id="c" * 32,
+        **values,
+    )
+
+
+def emit(reporter, **values):
+    return reporter.report(
+        **{
+            "predicted_peak_bytes": 100,
+            "observed_peak_bytes": 200,
+            "phase": "forward",
+            "replay_factory": lambda: {},
+            **values,
+        }
+    )
+
+
+def ledger(bound):
+    return json.loads((bound.spool_dir / ".retention.json").read_bytes())
+
+
+def test_reclamation_never_refunds_cumulative_budget(tmp_path):
+    bound = limits(tmp_path, max_reports=1)
+    reporter = reports.Reporter(5, spool_dir=tmp_path / "standalone")
+    with reports.report_retention_scope(bound):
+        path = emit(reporter)
+        assert path is not None and path.parent == bound.spool_dir
+        raw = path.read_bytes()
+        original = ledger(bound)
+        path.unlink()
+        assert emit(reporter) is None
+    assert not reporter.spool_dir.exists()
+    assert ledger(bound)["charges"] == original["charges"]
+    assert ledger(bound)["omitted"] == 1
+    assert ledger(bound)["omitted_bytes"] > 0
+    # Retransmission of the same captured bytes spends nothing new, even if
+    # recreating a previously reclaimed source copy for the original report.
+    assert (
+        reports.persist_report(raw, bound.spool_dir, retention=bound).read_bytes()
+        == raw
+    )
+    assert ledger(bound)["charges"] == original["charges"]
+
+
+def test_origin_scope_is_nested_and_exception_safe(tmp_path):
+    a = limits(tmp_path)
+    b = replace(a, spool_dir=tmp_path / "other", allowance_id="d" * 32)
+    reporter = reports.Reporter(5, spool_dir=tmp_path / "standalone")
+    with reports.report_retention_scope(a):
+        with pytest.raises(RuntimeError):
+            with reports.report_retention_scope(b):
+                assert emit(reporter).parent == b.spool_dir
+                raise RuntimeError("science")
+        assert emit(reporter).parent == a.spool_dir
+    assert emit(reporter).parent == reporter.spool_dir
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("producer_id", "d" * 32),
+        ("execution_id", "e" * 32),
+        ("allowance_id", "f" * 32),
+        ("max_bytes", 2 * 1024 * 1024),
+        ("max_reports", 5),
+    ],
+)
+def test_enrollment_cannot_be_replaced_or_replenished(tmp_path, field, value):
+    bound = limits(tmp_path)
+    reporter = reports.Reporter(5)
+    with reports.report_retention_scope(bound):
+        assert emit(reporter) is not None
+    original = (bound.spool_dir / ".retention.json").read_bytes()
+    with reports.report_retention_scope(replace(bound, **{field: value})):
+        assert emit(reporter) is None
+    assert (bound.spool_dir / ".retention.json").read_bytes() == original
+
+
+def test_ambiguous_payload_write_keeps_charge(tmp_path, monkeypatch):
+    bound = limits(tmp_path, max_reports=1)
+    reporter = reports.Reporter(5)
+
+    def fail(*args):
+        raise OSError("synthetic failure after charge")
+
+    with reports.report_retention_scope(bound):
+        with monkeypatch.context() as patch:
+            patch.setattr(reports.os, "link", fail)
+            assert emit(reporter) is None
+        assert len(ledger(bound)["charges"]) == 1
+        assert emit(reporter) is None
+    assert not list(bound.spool_dir.glob("[0-9a-f]*.json"))
+
+    assert ledger(bound)["omitted"] == 2
+    assert ledger(bound)["omitted_bytes"] > 0
+    assert reporter.failures == 2
+
+
+def test_planning_flood_keeps_oom_headroom_after_reclaim(tmp_path, monkeypatch):
+    monkeypatch.setattr(reports, "MAX_PLANNING_REPORTS", 1)
+    bound = limits(tmp_path)
+    reporter = reports.Reporter(5)
+    planning = {
+        "event": "planning_error",
+        "phase": "planning",
+        "observed_peak_bytes": None,
+    }
+    with reports.report_retention_scope(bound):
+        first = emit(reporter, **planning)
+        assert first is not None
+        first.unlink()
+        assert emit(reporter, **planning) is None
+        assert (
+            emit(reporter, oom=True, observed_peak_bytes=None, partial_peak_bytes=10)
+            is not None
+        )
+    assert len(ledger(bound)["charges"]) == 2
+
+
+@pytest.mark.parametrize(
+    "bound_values", [{"max_bytes": 0}, {"max_reports": 0}, {"max_bytes": 1}]
+)
+def test_exhausted_allowance_has_no_fallback(tmp_path, bound_values):
+    bound = limits(tmp_path, **bound_values)
+    reporter = reports.Reporter(5, spool_dir=tmp_path / "standalone")
+    with reports.report_retention_scope(bound):
+        assert emit(reporter) is None
+    assert not reporter.spool_dir.exists()
+    assert not list(bound.spool_dir.glob("[0-9a-f]*.json"))
+    assert ledger(bound)["omitted"] == 1
+    assert ledger(bound)["omitted_bytes"] > 0
+
+
+def test_unaccounted_spool_and_corrupt_ledger_refuse(tmp_path):
+    bound = limits(tmp_path)
+    bound.spool_dir.mkdir(mode=0o700)
+    foreign = bound.spool_dir / "foreign.json"
+    foreign.write_text("retained")
+    reporter = reports.Reporter(5)
+    with reports.report_retention_scope(bound):
+        assert emit(reporter) is None
+        assert foreign.read_text() == "retained"
+        foreign.unlink()
+        assert emit(reporter) is not None
+        (bound.spool_dir / ".retention.json").write_text("partial")
+        assert emit(reporter) is None
+
+
+def test_disabled_reporting_never_creates_assigned_spool(tmp_path):
+    bound = limits(tmp_path)
+    with reports.report_retention_scope(bound):
+        assert emit(reports.Reporter()) is None
+    assert not bound.spool_dir.exists()
+
+
+@pytest.mark.parametrize("bounded_by", ["bytes", "reports"])
+def test_interrupted_payload_is_not_duplicated_by_retry(tmp_path, bounded_by):
+    seed = emit(reports.Reporter(5, spool_dir=tmp_path / "seed"))
+    assert seed is not None
+    raw = seed.read_bytes()
+    bound = limits(
+        tmp_path,
+        **({"max_bytes": len(raw)} if bounded_by == "bytes" else {"max_reports": 1}),
+    )
+    committed = reports.persist_report(raw, bound.spool_dir, retention=bound)
+    charge = ledger(bound)
+    # Same durable state as death after payload fsync but before final link.
+    orphan = bound.spool_dir / ".pending-interrupted"
+    committed.rename(orphan)
+    with pytest.raises(ValueError, match="spool is full"):
+        reports.persist_report(raw, bound.spool_dir, retention=bound)
+    assert orphan.read_bytes() == raw
+    assert not committed.exists()
+    assert ledger(bound)["charges"] == charge["charges"]
+    assert ledger(bound)["omitted"] == charge["omitted"] + 1
+    assert ledger(bound)["omitted_bytes"] == charge["omitted_bytes"] + len(raw)
+
+
+def test_post_charge_refusal_counts_omission_without_refund(tmp_path):
+    bound = limits(tmp_path)
+    reporter = reports.Reporter(5)
+    with reports.report_retention_scope(bound):
+        assert emit(reporter) is not None
+        original = ledger(bound)
+        orphan = bound.spool_dir / ".pending-orphan"
+        orphan.write_bytes(b"x" * bound.max_bytes)
+        assert emit(reporter) is None
+    checked = ledger(bound)
+    assert len(checked["charges"]) == len(original["charges"]) + 1
+    assert checked["omitted"] == 1
+    assert checked["omitted_bytes"] > 0
+    assert orphan.stat().st_size == bound.max_bytes
+
+
+def test_omission_write_failure_preserves_original_failure(tmp_path, monkeypatch):
+    bound = limits(tmp_path)
+    seed = emit(reports.Reporter(5, spool_dir=tmp_path / "seed"))
+    assert seed is not None
+    raw = seed.read_bytes()
+    original_write = reports._planner_retention._write
+    calls = []
+
+    def write(path, value):
+        calls.append(None)
+        if len(calls) == 2:
+            raise OSError("omission ledger unavailable")
+        original_write(path, value)
+
+    def fail(*args):
+        raise ValueError("payload failed")
+
+    monkeypatch.setattr(reports._planner_retention, "_write", write)
+    monkeypatch.setattr(reports.os, "link", fail)
+    with pytest.raises(ValueError, match="payload failed"):
+        reports.persist_report(raw, bound.spool_dir, retention=bound)
+    assert len(calls) == 2
+    assert len(ledger(bound)["charges"]) == 1
+
+
+@pytest.mark.parametrize("assigned", [False, True])
+@pytest.mark.parametrize("failed_barrier", ["file", "directory"])
+def test_duplicate_retry_reestablishes_durability(
+    tmp_path, monkeypatch, assigned, failed_barrier
+):
+    seed = emit(reports.Reporter(5, spool_dir=tmp_path / "seed"))
+    assert seed is not None
+    raw = seed.read_bytes()
+    bound = limits(tmp_path)
+    path = bound.spool_dir / seed.name
+    retention = bound if assigned else None
+    fsync = reports.os.fsync
+
+    def fail_after_link(fd):
+        if path.exists() and stat.S_ISDIR(reports.os.fstat(fd).st_mode):
+            raise OSError("directory barrier unavailable")
+        fsync(fd)
+
+    monkeypatch.setattr(reports.os, "fsync", fail_after_link)
+    with pytest.raises(OSError, match="directory barrier unavailable"):
+        reports.persist_report(raw, bound.spool_dir, retention=retention)
+    assert path.read_bytes() == raw  # Visible is not proof of durable custody.
+    charges = ledger(bound)["charges"] if assigned else None
+    barriers = []
+
+    def retry_barrier(fd):
+        info = reports.os.fstat(fd)
+        if stat.S_ISDIR(info.st_mode):
+            barrier = "directory"
+        elif info.st_ino == path.stat().st_ino:
+            barrier = "file"
+        else:
+            return fsync(fd)
+        barriers.append(barrier)
+        if barrier == failed_barrier:
+            raise OSError("retry barrier unavailable")
+        fsync(fd)
+
+    monkeypatch.setattr(reports.os, "fsync", retry_barrier)
+    with pytest.raises(OSError, match="retry barrier unavailable"):
+        reports.persist_report(raw, bound.spool_dir, retention=retention)
+    assert failed_barrier in barriers
+    failed_barrier = None
+    barriers.clear()
+    assert reports.persist_report(raw, bound.spool_dir, retention=retention) == path
+    assert barriers == ["file", "directory"]
+    assert path.read_bytes() == raw
+    if assigned:
+        assert ledger(bound)["charges"] == charges  # Never charge a retry twice.
+
+
+@pytest.mark.parametrize("assigned", [False, True])
+def test_capture_off_suppresses_constructor_before_replay_or_io(
+    tmp_path, monkeypatch, assigned
+):
+    bound = limits(tmp_path, max_bytes=0, max_reports=0) if assigned else None
+    replayed = []
+    sourced = []
+    monkeypatch.setattr(reports, "_source_files", lambda: sourced.append(True))
+
+    class Constructed:
+        def __init__(self):
+            self.reporter = reports.Reporter(5, spool_dir=tmp_path / "standalone")
+            self.path = emit(
+                self.reporter, replay_factory=lambda: replayed.append(True)
+            )
+
+    with reports.report_retention_scope(bound, capture=False):
+        constructed = Constructed()
+    assert constructed.path is None
+    assert constructed.reporter.failures == 0
+    assert replayed == sourced == []
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_capture_suppression_nests_and_restores_after_exception(tmp_path):
+    reporter = reports.Reporter(5, spool_dir=tmp_path / "standalone")
+    bound = limits(tmp_path)
+    with pytest.raises(RuntimeError, match="science"):
+        with reports.report_retention_scope(None, capture=False):
+            with reports.report_retention_scope(bound):
+                assert emit(reporter) is None
+            raise RuntimeError("science")
+    assert not bound.spool_dir.exists()
+    with reports.report_retention_scope(bound):
+        with reports.report_retention_scope(None, capture=False):
+            assert emit(reporter) is None
+        assert emit(reporter).parent == bound.spool_dir
+    assert emit(reporter).parent == reporter.spool_dir
+
+
+def test_capture_off_differs_from_zero_allowance(tmp_path):
+    bound = limits(tmp_path, max_bytes=0, max_reports=0)
+    reporter = reports.Reporter(5)
+    with reports.report_retention_scope(bound, capture=False):
+        assert emit(reporter) is None
+    assert not bound.spool_dir.exists()
+    with reports.report_retention_scope(bound):
+        assert emit(reporter) is None
+    assert ledger(bound)["omitted"] == 1
+
+
+def test_capture_scope_inherits_into_to_thread(tmp_path):
+    bound = limits(tmp_path)
+    reporter = reports.Reporter(5, spool_dir=tmp_path / "standalone")
+
+    async def run():
+        with reports.report_retention_scope(bound, capture=False):
+            assert await asyncio.to_thread(emit, reporter) is None
+        assert not bound.spool_dir.exists()
+        with reports.report_retention_scope(bound):
+            assert (await asyncio.to_thread(emit, reporter)).parent == bound.spool_dir
+
+    asyncio.run(run())
+    assert not reporter.spool_dir.exists()
+
+
+@pytest.mark.parametrize("capture", [None, 0, "false"])
+def test_capture_flag_refuses_non_boolean_without_changing_scope(tmp_path, capture):
+    with pytest.raises(ValueError, match="capture must be a boolean"):
+        with reports.report_retention_scope(None, capture=capture):
+            raise AssertionError("invalid capture flag entered")
+    assert emit(reports.Reporter(5, spool_dir=tmp_path / "standalone")) is not None