diff --git a/pg_llm_batch/result_application.py b/pg_llm_batch/result_application.py
new file mode 100644
index 000000000..631263cdd
--- /dev/null
+++ b/pg_llm_batch/result_application.py
@@ -0,0 +1,317 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) ContextualWisdomLab.
+"""Atomic local application of streamed provider results with checkpoints.
+
+The helper in this module deliberately owns no PostgreSQL connection and no
+transaction lifecycle. A caller supplies a cursor that already belongs to the
+transaction in which both the local business effect and durable checkpoint
+advance must occur. The business callback receives a package-scoped,
+same-thread cursor capability rather than the raw cursor. This permits atomicity
+only for effects executed synchronously through that capability; it does not
+create a distributed exactly-once guarantee for external APIs, queues, object
+stores, other databases, or independently retained caller resources.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+from concurrent.futures import Future as ConcurrentFuture
+from dataclasses import dataclass
+from threading import get_ident
+from typing import Any, Callable, Mapping
+
+from .checkpoint_store import CheckpointConflictError
+from .exceptions import PgLlmBatchError, ValidationError
+from .result_streaming import BatchResultCheckpoint, CheckpointedBatchResultRecord
+
+
+class ResultApplicationError(PgLlmBatchError):
+    """Report one bounded failure while applying a checkpointed result."""
+
+    def __init__(self, phase: str) -> None:
+        """Create fixed diagnostic evidence for one application phase."""
+        super().__init__(
+            message="Checkpointed result application failed",
+            error_code="RESULT_APPLICATION_ERROR",
+            details={"phase": phase},
+        )
+
+
+@dataclass(frozen=True)
+class ResultApplicationOutcome:
+    """Describe whether one local record effect was newly applied."""
+
+    applied: bool
+    checkpoint: BatchResultCheckpoint
+
+
+class _ResultApplicationCursor:
+    """Expose a revocable same-thread subset of caller transaction authority.
+
+    The facade intentionally does not expose the underlying connection, commit,
+    rollback, copy, streaming, or arbitrary attribute access. Synchronous record
+    effects may execute statements and consume ordinary cursor results while the
+    callback is active. The capability is revoked as soon as the callback
+    returns or raises, and use from any other thread fails before the raw cursor
+    is touched.
+    """
+
+    __slots__ = ("__active", "__cursor", "__owner_thread_id")
+
+    def __init__(self, cursor: Any) -> None:
+        """Bind one raw cursor to the constructing thread for one callback."""
+        self.__cursor = cursor
+        self.__owner_thread_id = get_ident()
+        self.__active = True
+
+    def _revoke(self) -> None:
+        """Remove package-supplied cursor authority after callback completion."""
+        self.__active = False
+
+    def _assert_usable(self) -> None:
+        """Reject expired or cross-thread use with bounded package evidence."""
+        if not self.__active or get_ident() != self.__owner_thread_id:
+            raise ResultApplicationError("record_effect") from None
+
+    def execute(self, *args: Any, **kwargs: Any) -> _ResultApplicationCursor:
+        """Execute one statement synchronously without returning the raw cursor."""
+        self._assert_usable()
+        self.__cursor.execute(*args, **kwargs)
+        return self
+
+    def executemany(self, *args: Any, **kwargs: Any) -> _ResultApplicationCursor:
+        """Execute one parameter sequence without returning the raw cursor."""
+        self._assert_usable()
+        self.__cursor.executemany(*args, **kwargs)
+        return self
+
+    def fetchone(self, *args: Any, **kwargs: Any) -> Any:
+        """Fetch one result while this callback owns the scoped capability."""
+        self._assert_usable()
+        return self.__cursor.fetchone(*args, **kwargs)
+
+    def fetchmany(self, *args: Any, **kwargs: Any) -> Any:
+        """Fetch a bounded result page while the scoped capability is active."""
+        self._assert_usable()
+        return self.__cursor.fetchmany(*args, **kwargs)
+
+    def fetchall(self, *args: Any, **kwargs: Any) -> Any:
+        """Fetch remaining results while the scoped capability is active."""
+        self._assert_usable()
+        return self.__cursor.fetchall(*args, **kwargs)
+
+
+def _redacted_validation_error(field: str, reason: str) -> ValidationError:
+    """Build a validation error without retaining caller-controlled content."""
+    return ValidationError(field=field, value="<redacted>", reason=reason)
+
+
+def _checkpoint_primitive_type_error(checkpoint: BatchResultCheckpoint) -> str | None:
+    """Return the first checkpoint field whose primitive type can execute behavior."""
+    for field in (
+        "batch_id",
+        "endpoint_alias",
+        "file_kind",
+        "file_id",
+        "prefix_sha256",
+    ):
+        if type(getattr(checkpoint, field)) is not str:
+            return field
+    for field in (
+        "schema_version",
+        "file_line_number",
+        "batch_line_count",
+        "record_count",
+    ):
+        if type(getattr(checkpoint, field)) is not int:
+            return field
+    return None
+
+
+def _validate_item_and_effect(
+    item: Any,
+    apply_record: Any,
+) -> CheckpointedBatchResultRecord:
+    """Validate the local application boundary before store or callback work."""
+    if type(item) is not CheckpointedBatchResultRecord:
+        raise _redacted_validation_error(
+            "item", "must be an exact checkpointed batch result record"
+        )
+    checkpoint = item.checkpoint
+    if type(checkpoint) is not BatchResultCheckpoint:
+        raise _redacted_validation_error(
+            "item.checkpoint", "must be an exact batch result checkpoint"
+        )
+    checkpoint_field = _checkpoint_primitive_type_error(checkpoint)
+    if checkpoint_field is not None:
+        raise _redacted_validation_error(
+            f"item.checkpoint.{checkpoint_field}",
+            "must use an exact built-in primitive type",
+        )
+    if type(item.batch_id) is not str:
+        raise _redacted_validation_error(
+            "item.batch_id", "must be an exact built-in string"
+        )
+    if type(item.file_kind) is not str:
+        raise _redacted_validation_error(
+            "item.file_kind", "must be an exact built-in string"
+        )
+    if not callable(apply_record):
+        raise _redacted_validation_error("apply_record", "must be callable")
+    static_call = inspect.getattr_static(apply_record, "__call__", None)
+    if isinstance(static_call, (staticmethod, classmethod)):
+        static_call = static_call.__func__
+    if inspect.iscoroutinefunction(apply_record) or inspect.iscoroutinefunction(
+        static_call
+    ):
+        raise _redacted_validation_error(
+            "apply_record", "must complete synchronously in the caller transaction"
+        )
+    if item.batch_id != checkpoint.batch_id:
+        raise _redacted_validation_error(
+            "item.batch_id", "must match the checkpoint batch identity"
+        )
+    if item.file_kind != checkpoint.file_kind:
+        raise _redacted_validation_error(
+            "item.file_kind", "must match the checkpoint file kind"
+        )
+    if type(item.record) is not dict:
+        raise _redacted_validation_error("item.record", "must be an exact JSON object")
+    return item
+
+
+def apply_checkpointed_result_in_transaction(
+    cursor: Any,
+    checkpoint_store: Any,
+    consumer_name: str,
+    item: CheckpointedBatchResultRecord,
+    apply_record: Callable[[Any, Mapping[str, Any]], None],
+) -> ResultApplicationOutcome:
+    """Apply one result and advance its checkpoint in the caller's transaction.
+
+    The item, checkpoint, checkpoint primitive fields, JSON object, loaded
+    predecessor, and save confirmation must use exact package-owned or built-in
+    types. Subclasses are rejected before their behavior-bearing comparison or
+    attribute hooks can execute, so caller-controlled subclass code cannot
+    disclose diagnostics or forge durable confirmation.
+
+    The durable predecessor is loaded and validated before the local effect. An
+    exact replay returns without re-running the effect, while a count regression
+    is rejected before caller-owned business logic. Fresh work invokes
+    ``apply_record`` with a package-scoped cursor facade on the supplied
+    transaction and advances the checkpoint only after that callback completes
+    synchronously and returns ``None``. The facade permits ordinary synchronous
+    ``execute``/``executemany`` and ``fetch*`` operations only on the callback's
+    original thread. It is revoked on every callback exit, so deferred work
+    cannot retain package-supplied transaction cursor authority after return.
+    This is an authority boundary, not a claim that Python can forcibly
+    terminate arbitrary already-running Futures, Tasks, threads, or other
+    caller-retained resources.
+
+    Statically visible asynchronous callables, including static-method and
+    class-method descriptors, are rejected before checkpoint-store access. A raw
+    coroutine returned by an otherwise synchronous callable is closed, and
+    returned pending :class:`asyncio.Future` or
+    :class:`concurrent.futures.Future` work receives best-effort cancellation
+    after the scoped cursor has already been revoked. Any non-``None`` return is
+    rejected as a record-effect failure. The checkpoint store must then confirm
+    the exact requested checkpoint before success is reported. The caller
+    remains responsible for committing or rolling back the surrounding
+    transaction.
+
+    ``CheckpointConflictError`` is intentionally preserved as the stable retry
+    signal from both checkpoint load and save operations. All other
+    store/callback failures are replaced with a fixed phase-only package error
+    after their exception scope has ended, preventing implicit traceback context
+    from retaining provider or database diagnostics.
+    """
+    candidate = _validate_item_and_effect(item, apply_record)
+
+    load_failure: ResultApplicationError | None = None
+    previous: BatchResultCheckpoint | None = None
+    try:
+        previous = checkpoint_store.load_in_transaction(
+            cursor,
+            consumer_name,
+            candidate.batch_id,
+            candidate.checkpoint.endpoint_alias,
+        )
+    except CheckpointConflictError:
+        raise
+    except Exception:
+        load_failure = ResultApplicationError("checkpoint_load")
+    if load_failure is not None:
+        raise load_failure from None
+    if previous is not None:
+        if type(previous) is not BatchResultCheckpoint:
+            raise ResultApplicationError("checkpoint_load") from None
+        if _checkpoint_primitive_type_error(previous) is not None:
+            raise ResultApplicationError("checkpoint_load") from None
+        if (
+            previous.batch_id != candidate.checkpoint.batch_id
+            or previous.endpoint_alias != candidate.checkpoint.endpoint_alias
+            or previous.file_kind != candidate.checkpoint.file_kind
+            or previous.file_id != candidate.checkpoint.file_id
+        ):
+            raise ResultApplicationError("checkpoint_load") from None
+
+    if previous == candidate.checkpoint:
+        return ResultApplicationOutcome(applied=False, checkpoint=candidate.checkpoint)
+    if previous is not None and (
+        candidate.checkpoint.record_count <= previous.record_count
+        or candidate.checkpoint.batch_line_count <= previous.batch_line_count
+    ):
+        raise CheckpointConflictError(
+            consumer_name,
+            candidate.batch_id,
+            "checkpoint_regression",
+        ) from None
+
+    effect_failure: ResultApplicationError | None = None
+    effect_cursor = _ResultApplicationCursor(cursor)
+    try:
+        try:
+            effect_result = apply_record(effect_cursor, candidate.record)
+        finally:
+            effect_cursor._revoke()
+        if inspect.iscoroutine(effect_result):
+            effect_result.close()
+        elif isinstance(effect_result, (asyncio.Future, ConcurrentFuture)):
+            effect_result.cancel()
+        if effect_result is not None:
+            effect_failure = ResultApplicationError("record_effect")
+    except Exception:
+        effect_failure = ResultApplicationError("record_effect")
+    if effect_failure is not None:
+        raise effect_failure from None
+
+    save_failure: ResultApplicationError | None = None
+    try:
+        saved_checkpoint = checkpoint_store.save_in_transaction(
+            cursor,
+            consumer_name,
+            candidate.checkpoint,
+            expected_previous=previous,
+        )
+        if type(saved_checkpoint) is not BatchResultCheckpoint:
+            save_failure = ResultApplicationError("checkpoint_save")
+        elif _checkpoint_primitive_type_error(saved_checkpoint) is not None:
+            save_failure = ResultApplicationError("checkpoint_save")
+        elif saved_checkpoint != candidate.checkpoint:
+            save_failure = ResultApplicationError("checkpoint_save")
+    except CheckpointConflictError:
+        raise
+    except Exception:
+        save_failure = ResultApplicationError("checkpoint_save")
+    if save_failure is not None:
+        raise save_failure from None
+
+    return ResultApplicationOutcome(applied=True, checkpoint=candidate.checkpoint)
+
+
+__all__ = [
+    "ResultApplicationError",
+    "ResultApplicationOutcome",
+    "apply_checkpointed_result_in_transaction",
+]
diff --git a/tests/test_result_application.py b/tests/test_result_application.py
new file mode 100644
index 000000000..ab3fc7036
--- /dev/null
+++ b/tests/test_result_application.py
@@ -0,0 +1,389 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Regression tests for transactional provider-result application."""
+
+from __future__ import annotations
+
+import traceback
+from dataclasses import replace
+from typing import Any, Callable
+
+import pytest
+
+from pg_llm_batch.checkpoint_store import CheckpointConflictError
+from pg_llm_batch.exceptions import ValidationError
+from pg_llm_batch.result_application import (
+    ResultApplicationError,
+    ResultApplicationOutcome,
+    apply_checkpointed_result_in_transaction,
+)
+from pg_llm_batch.result_streaming import (
+    BatchResultCheckpoint,
+    CheckpointedBatchResultRecord,
+)
+
+
+def _checkpoint(*, record_count: int = 1, digest: str = "a" * 64) -> BatchResultCheckpoint:
+    """Build one valid checkpoint for the focused transaction tests."""
+    return BatchResultCheckpoint(
+        schema_version=1,
+        batch_id="batch-123",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=record_count,
+        batch_line_count=record_count,
+        record_count=record_count,
+        prefix_sha256=digest,
+    )
+
+
+def _item(checkpoint: BatchResultCheckpoint) -> CheckpointedBatchResultRecord:
+    """Pair one decoded record with its exact checkpoint identity."""
+    return CheckpointedBatchResultRecord(
+        batch_id=checkpoint.batch_id,
+        file_kind=checkpoint.file_kind,
+        record={"custom_id": f"request-{checkpoint.record_count}"},
+        checkpoint=checkpoint,
+    )
+
+
+class _Store:
+    """Minimal caller-transaction checkpoint store recording operation order."""
+
+    def __init__(self, previous: BatchResultCheckpoint | None = None) -> None:
+        self.previous = previous
+        self.events: list[tuple[Any, ...]] = []
+        self.load_error: Exception | None = None
+        self.save_error: Exception | None = None
+
+    def load_in_transaction(
+        self,
+        cursor: Any,
+        consumer_name: str,
+        batch_id: str,
+        endpoint_alias: str,
+    ) -> BatchResultCheckpoint | None:
+        """Return configured durable state or one injected failure."""
+        self.events.append(("load", cursor, consumer_name, batch_id, endpoint_alias))
+        if self.load_error is not None:
+            raise self.load_error
+        return self.previous
+
+    def save_in_transaction(
+        self,
+        cursor: Any,
+        consumer_name: str,
+        checkpoint: BatchResultCheckpoint,
+        *,
+        expected_previous: BatchResultCheckpoint | None = None,
+    ) -> BatchResultCheckpoint:
+        """Record compare-and-swap inputs or one injected failure."""
+        self.events.append(
+            ("save", cursor, consumer_name, checkpoint, expected_previous)
+        )
+        if self.save_error is not None:
+            raise self.save_error
+        self.previous = checkpoint
+        return checkpoint
+
+
+def test_applies_local_effect_before_checkpoint_in_same_caller_transaction() -> None:
+    """Fresh work must execute its local effect then advance the exact checkpoint."""
+    cursor = object()
+    checkpoint = _checkpoint()
+    item = _item(checkpoint)
+    store = _Store()
+
+    def effect(seen_cursor: Any, record: dict[str, Any]) -> None:
+        store.events.append(("effect", seen_cursor, record.copy()))
+
+    outcome = apply_checkpointed_result_in_transaction(
+        cursor,
+        store,
+        "result-writer",
+        item,
+        effect,
+    )
+
+    assert outcome == ResultApplicationOutcome(applied=True, checkpoint=checkpoint)
+    assert [event[0] for event in store.events] == ["load", "effect", "save"]
+    assert store.events[0][1] is cursor
+    assert store.events[1][1] is not cursor
+    assert store.events[1][2] == item.record
+    assert store.events[2][1] is cursor
+    assert store.events[2][-1] is None
+
+
+def test_existing_checkpoint_is_supplied_as_compare_and_swap_predecessor() -> None:
+    """Advancement must bind the exact previously loaded durable checkpoint."""
+    previous = _checkpoint(record_count=1, digest="a" * 64)
+    checkpoint = _checkpoint(record_count=2, digest="b" * 64)
+    store = _Store(previous)
+
+    outcome = apply_checkpointed_result_in_transaction(
+        object(),
+        store,
+        "result-writer",
+        _item(checkpoint),
+        lambda _cursor, _record: None,
+    )
+
+    assert outcome.applied is True
+    assert store.events[-1][-1] == previous
+
+
+def test_exact_checkpoint_replay_is_idempotent_without_reapplying_effect() -> None:
+    """An already acknowledged record must not repeat its local business effect."""
+    checkpoint = _checkpoint()
+    store = _Store(checkpoint)
+    called = False
+
+    def effect(_cursor: Any, _record: dict[str, Any]) -> None:
+        nonlocal called
+        called = True
+
+    outcome = apply_checkpointed_result_in_transaction(
+        object(),
+        store,
+        "result-writer",
+        _item(checkpoint),
+        effect,
+    )
+
+    assert outcome == ResultApplicationOutcome(applied=False, checkpoint=checkpoint)
+    assert called is False
+    assert [event[0] for event in store.events] == ["load"]
+
+
+@pytest.mark.parametrize(
+    ("mutation", "field"),
+    [
+        (
+            lambda item: CheckpointedBatchResultRecord(
+                batch_id="other-batch",
+                file_kind=item.file_kind,
+                record=item.record,
+                checkpoint=item.checkpoint,
+            ),
+            "item.batch_id",
+        ),
+        (
+            lambda item: CheckpointedBatchResultRecord(
+                batch_id=item.batch_id,
+                file_kind="error",
+                record=item.record,
+                checkpoint=item.checkpoint,
+            ),
+            "item.file_kind",
+        ),
+        (
+            lambda item: CheckpointedBatchResultRecord(
+                batch_id=item.batch_id,
+                file_kind=item.file_kind,
+                record=[],  # type: ignore[arg-type]
+                checkpoint=item.checkpoint,
+            ),
+            "item.record",
+        ),
+    ],
+)
+def test_record_identity_is_validated_before_store_or_effect(
+    mutation: Callable[[CheckpointedBatchResultRecord], CheckpointedBatchResultRecord],
+    field: str,
+) -> None:
+    """Decoded payload identity must agree with its checkpoint before side effects."""
+    store = _Store()
+    item = mutation(_item(_checkpoint()))
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", item, lambda _cursor, _record: None
+        )
+
+    assert caught.value.details["field"] == field
+    assert caught.value.details["value"] == "<redacted>"
+    assert store.events == []
+
+
+def test_argument_types_fail_closed_before_store_access() -> None:
+    """The helper must reject unsupported records and non-callable effect hooks."""
+    store = _Store()
+    with pytest.raises(ValidationError) as bad_item:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", object(), lambda _cursor, _record: None
+        )
+    assert bad_item.value.details["field"] == "item"
+
+    with pytest.raises(ValidationError) as bad_effect:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), None  # type: ignore[arg-type]
+        )
+    assert bad_effect.value.details["field"] == "apply_record"
+    assert store.events == []
+
+
+def test_async_effect_hook_fails_before_checkpoint_store_access() -> None:
+    """An async callback must not be mistaken for an executed synchronous effect."""
+    store = _Store()
+
+    async def async_effect(_cursor: Any, _record: dict[str, Any]) -> None:
+        return None
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), async_effect
+        )
+
+    assert caught.value.details["field"] == "apply_record"
+    assert caught.value.details["value"] == "<redacted>"
+    assert store.events == []
+
+
+def test_effect_must_complete_synchronously_before_checkpoint_advance() -> None:
+    """A callback return value must fail closed instead of acknowledging deferred work."""
+    store = _Store()
+
+    def effect(cursor: Any, record: dict[str, Any]) -> object:
+        store.events.append(("effect", cursor, record.copy()))
+        return object()
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), effect
+        )
+
+    assert caught.value.details == {"phase": "record_effect"}
+    assert [event[0] for event in store.events] == ["load", "effect"]
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+
+
+def test_effect_failure_is_bounded_and_never_advances_checkpoint() -> None:
+    """Sensitive callback diagnostics must not replace the stable package error."""
+    store = _Store()
+
+    def effect(_cursor: Any, _record: dict[str, Any]) -> None:
+        raise RuntimeError("SECRET-SENTINEL provider payload diagnostic")
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), effect
+        )
+
+    rendered = "".join(
+        traceback.format_exception(type(caught.value), caught.value, caught.value.__traceback__)
+    )
+    assert caught.value.details == {"phase": "record_effect"}
+    assert "SECRET-SENTINEL" not in rendered
+    assert [event[0] for event in store.events] == ["load"]
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+
+
+@pytest.mark.parametrize("phase", ["checkpoint_load", "checkpoint_save"])
+def test_unexpected_store_failure_is_bounded_without_database_diagnostics(
+    phase: str,
+) -> None:
+    """Unexpected store diagnostics must be replaced by finite phase evidence."""
+    store = _Store()
+    error = RuntimeError("SECRET-SENTINEL database diagnostic")
+    if phase == "checkpoint_load":
+        store.load_error = error
+    else:
+        store.save_error = error
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), lambda _c, _r: None
+        )
+
+    rendered = "".join(
+        traceback.format_exception(type(caught.value), caught.value, caught.value.__traceback__)
+    )
+    assert caught.value.details == {"phase": phase}
+    assert "SECRET-SENTINEL" not in rendered
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+
+
+def test_checkpoint_conflict_remains_a_stable_retry_signal() -> None:
+    """Known compare-and-swap conflicts must retain their package-owned contract."""
+    store = _Store()
+    store.save_error = CheckpointConflictError(
+        "result-writer", "batch-123", "expected_previous_stale"
+    )
+
+    with pytest.raises(CheckpointConflictError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), lambda _c, _r: None
+        )
+
+    assert caught.value.reason == "expected_previous_stale"
+
+
+def test_checkpoint_load_conflict_remains_a_stable_retry_signal() -> None:
+    """Load-time CAS conflicts must retain their package-owned retry contract."""
+    store = _Store()
+    store.load_error = CheckpointConflictError(
+        "result-writer", "batch-123", "load_snapshot_stale"
+    )
+
+    with pytest.raises(CheckpointConflictError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), lambda _c, _r: None
+        )
+
+    assert caught.value.reason == "load_snapshot_stale"
+    assert [event[0] for event in store.events] == ["load"]
+
+
+def test_malformed_loaded_checkpoint_fails_before_record_effect() -> None:
+    """Malformed durable predecessor evidence must fail before local effects."""
+    store = _Store()
+    store.previous = object()  # type: ignore[assignment]
+    effect_called = False
+
+    def effect(_cursor: Any, _record: dict[str, Any]) -> None:
+        nonlocal effect_called
+        effect_called = True
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), effect
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_load"}
+    assert effect_called is False
+    assert [event[0] for event in store.events] == ["load"]
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+
+
+@pytest.mark.parametrize(
+    "previous",
+    [
+        replace(_checkpoint(), batch_id="other-batch"),
+        replace(_checkpoint(), endpoint_alias="other-endpoint"),
+    ],
+)
+def test_loaded_checkpoint_identity_must_match_requested_stream_before_effect(
+    previous: BatchResultCheckpoint,
+) -> None:
+    """A cross-stream predecessor must fail before caller-owned business effects."""
+    store = _Store(previous)
+    effect_called = False
+
+    def effect(_cursor: Any, _record: dict[str, Any]) -> None:
+        nonlocal effect_called
+        effect_called = True
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(_checkpoint()), effect
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_load"}
+    assert effect_called is False
+    assert [event[0] for event in store.events] == ["load"]
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
diff --git a/tests/test_result_application_async_callable.py b/tests/test_result_application_async_callable.py
new file mode 100644
index 000000000..468e3ba54
--- /dev/null
+++ b/tests/test_result_application_async_callable.py
@@ -0,0 +1,401 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Regressions for asynchronous result-effect boundaries."""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from concurrent.futures import Future as ConcurrentFuture
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
+
+import pytest
+
+from pg_llm_batch.exceptions import ValidationError
+from pg_llm_batch.result_application import (
+    ResultApplicationError,
+    apply_checkpointed_result_in_transaction,
+)
+from pg_llm_batch.result_streaming import (
+    BatchResultCheckpoint,
+    CheckpointedBatchResultRecord,
+)
+
+
+class _RecordingStore:
+    """Record whether the application seam reached checkpoint storage."""
+
+    def __init__(self) -> None:
+        self.events: list[str] = []
+
+    def load_in_transaction(self, *_args: object) -> None:
+        """Record checkpoint loading and provide no durable predecessor."""
+        self.events.append("load")
+        return None
+
+    def save_in_transaction(self, *_args: object, **_kwargs: object) -> None:
+        """Record an unexpected checkpoint save after an invalid effect."""
+        self.events.append("save")
+        return None
+
+
+class _SuccessfulStore(_RecordingStore):
+    """Confirm the requested checkpoint after one successful local effect."""
+
+    def save_in_transaction(
+        self,
+        _cursor: object,
+        _consumer_name: str,
+        checkpoint: BatchResultCheckpoint,
+        **_kwargs: object,
+    ) -> BatchResultCheckpoint:
+        """Record checkpoint persistence and return exact confirmation."""
+        self.events.append("save")
+        return checkpoint
+
+
+class _AsyncCallableEffect:
+    """Represent an ordinary callable object with asynchronous invocation."""
+
+    async def __call__(self, _cursor: Any, _record: dict[str, Any]) -> None:
+        """Model deferred work that cannot share the caller transaction."""
+        return None
+
+
+class _StaticAsyncCallableEffect:
+    """Represent an asynchronous static-method callable object."""
+
+    @staticmethod
+    async def __call__(_cursor: Any, _record: dict[str, Any]) -> None:
+        """Model a descriptor-wrapped asynchronous effect."""
+        return None
+
+
+class _ClassAsyncCallableEffect:
+    """Represent an asynchronous class-method callable object."""
+
+    @classmethod
+    async def __call__(
+        cls,
+        _cursor: Any,
+        _record: dict[str, Any],
+    ) -> None:
+        """Model another descriptor-wrapped asynchronous effect."""
+        return None
+
+
+class _CoroutineReturningEffect:
+    """Return a raw coroutine from an otherwise synchronous callable."""
+
+    def __init__(self) -> None:
+        self.returned_coroutine: Any = None
+
+    def __call__(self, _cursor: Any, _record: dict[str, Any]) -> Any:
+        """Create deferred work whose frame retains caller-owned arguments."""
+
+        async def _deferred_work() -> None:
+            return None
+
+        self.returned_coroutine = _deferred_work()
+        return self.returned_coroutine
+
+
+class _FutureReturningEffect:
+    """Return scheduled-style deferred work from a synchronous callable."""
+
+    def __init__(self) -> None:
+        self.event_loop = asyncio.new_event_loop()
+        self.returned_future: asyncio.Future[None] = self.event_loop.create_future()
+
+    def __call__(self, _cursor: Any, _record: dict[str, Any]) -> asyncio.Future[None]:
+        """Return a pending future that must not outlive bounded rejection."""
+        return self.returned_future
+
+    def close(self) -> None:
+        """Release the isolated event loop used by this regression fixture."""
+        self.event_loop.close()
+
+
+class _ConcurrentFutureReturningEffect:
+    """Return a thread-pool-style future from a synchronous callable."""
+
+    def __init__(self) -> None:
+        self.returned_future: ConcurrentFuture[None] = ConcurrentFuture()
+
+    def __call__(self, _cursor: Any, _record: dict[str, Any]) -> ConcurrentFuture[None]:
+        """Return pending concurrent work that must receive cancellation."""
+        return self.returned_future
+
+
+class _ExecutableCursor:
+    """Record raw database operations performed through the scoped facade."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[Any, ...]] = []
+
+    def execute(self, *args: Any, **kwargs: Any) -> _ExecutableCursor:
+        """Record one statement execution and mimic Psycopg's cursor return."""
+        self.calls.append(("execute", args, kwargs))
+        return self
+
+    def executemany(self, *args: Any, **kwargs: Any) -> _ExecutableCursor:
+        """Record one many-parameter execution and mimic cursor return."""
+        self.calls.append(("executemany", args, kwargs))
+        return self
+
+    def fetchone(self) -> tuple[str]:
+        """Return one deterministic row."""
+        self.calls.append(("fetchone",))
+        return ("one",)
+
+    def fetchmany(self, size: int) -> list[tuple[str]]:
+        """Return one deterministic bounded page."""
+        self.calls.append(("fetchmany", size))
+        return [("many",)]
+
+    def fetchall(self) -> list[tuple[str]]:
+        """Return one deterministic remainder."""
+        self.calls.append(("fetchall",))
+        return [("all",)]
+
+
+class _RunningConcurrentFutureReturningEffect:
+    """Keep one future running until the application seam has rejected it."""
+
+    def __init__(self) -> None:
+        self.executor = ThreadPoolExecutor(max_workers=1)
+        self.started = threading.Event()
+        self.release = threading.Event()
+        self.returned_future: ConcurrentFuture[None] | None = None
+
+    def __call__(self, cursor: Any, _record: dict[str, Any]) -> ConcurrentFuture[None]:
+        """Return already-running work that attempts cursor use after rejection."""
+
+        def _deferred_work() -> None:
+            self.started.set()
+            self.release.wait(timeout=2)
+            cursor.execute("SELECT 1")
+
+        self.returned_future = self.executor.submit(_deferred_work)
+        assert self.started.wait(timeout=1)
+        return self.returned_future
+
+    def close(self) -> None:
+        """Release and join the worker even when the regression assertion fails."""
+        self.release.set()
+        self.executor.shutdown(wait=True)
+
+
+class _CrossThreadCursorEffect:
+    """Attempt cursor use from another thread before the callback returns."""
+
+    def __init__(self) -> None:
+        self.worker_failure: Exception | None = None
+
+    def __call__(self, cursor: Any, _record: dict[str, Any]) -> None:
+        """Prove the live capability remains restricted to its owner thread."""
+
+        def _worker() -> None:
+            try:
+                cursor.execute("SELECT cross_thread")
+            except Exception as exc:
+                self.worker_failure = exc
+
+        worker = threading.Thread(target=_worker)
+        worker.start()
+        worker.join(timeout=1)
+        assert not worker.is_alive()
+
+
+def _item() -> CheckpointedBatchResultRecord:
+    """Build one valid result item for the focused effect boundary."""
+    checkpoint = BatchResultCheckpoint(
+        schema_version=1,
+        batch_id="batch-123",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=1,
+        batch_line_count=1,
+        record_count=1,
+        prefix_sha256="a" * 64,
+    )
+    return CheckpointedBatchResultRecord(
+        batch_id=checkpoint.batch_id,
+        file_kind=checkpoint.file_kind,
+        record={"custom_id": "request-1"},
+        checkpoint=checkpoint,
+    )
+
+
+@pytest.mark.parametrize(
+    "effect_type",
+    [
+        _AsyncCallableEffect,
+        _StaticAsyncCallableEffect,
+        _ClassAsyncCallableEffect,
+    ],
+)
+def test_async_callable_object_fails_before_checkpoint_store_access(
+    effect_type: type[Any],
+) -> None:
+    """Every statically visible async ``__call__`` must fail before store work."""
+    store = _RecordingStore()
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(),
+            effect_type(),
+        )
+
+    assert caught.value.details["field"] == "apply_record"
+    assert caught.value.details["value"] == "<redacted>"
+    assert store.events == []
+
+
+def test_scoped_cursor_supports_sync_subset_then_revokes_on_return() -> None:
+    """Synchronous DB-API work must succeed without leaking raw cursor authority."""
+    store = _SuccessfulStore()
+    raw_cursor = _ExecutableCursor()
+    observed: dict[str, Any] = {}
+
+    def effect(cursor: Any, _record: dict[str, Any]) -> None:
+        observed["cursor"] = cursor
+        assert cursor is not raw_cursor
+        assert cursor.execute("SELECT 1", answer=42) is cursor
+        assert cursor.executemany("SELECT %s", [(1,), (2,)]) is cursor
+        assert cursor.fetchone() == ("one",)
+        assert cursor.fetchmany(1) == [("many",)]
+        assert cursor.fetchall() == [("all",)]
+
+    outcome = apply_checkpointed_result_in_transaction(
+        raw_cursor,
+        store,
+        "result-writer",
+        _item(),
+        effect,
+    )
+
+    assert outcome.applied is True
+    assert store.events == ["load", "save"]
+    assert [call[0] for call in raw_cursor.calls] == [
+        "execute",
+        "executemany",
+        "fetchone",
+        "fetchmany",
+        "fetchall",
+    ]
+    with pytest.raises(ResultApplicationError) as revoked:
+        observed["cursor"].execute("SELECT after_return")
+    assert revoked.value.details == {"phase": "record_effect"}
+    assert len(raw_cursor.calls) == 5
+
+
+def test_scoped_cursor_rejects_cross_thread_use_while_callback_is_active() -> None:
+    """A live scoped cursor must reject worker-thread use before raw I/O."""
+    store = _SuccessfulStore()
+    raw_cursor = _ExecutableCursor()
+    effect = _CrossThreadCursorEffect()
+
+    outcome = apply_checkpointed_result_in_transaction(
+        raw_cursor,
+        store,
+        "result-writer",
+        _item(),
+        effect,
+    )
+
+    assert outcome.applied is True
+    assert isinstance(effect.worker_failure, ResultApplicationError)
+    assert effect.worker_failure.details == {"phase": "record_effect"}
+    assert raw_cursor.calls == []
+    assert store.events == ["load", "save"]
+
+
+def test_returned_coroutine_is_closed_before_bounded_failure() -> None:
+    """Rejected deferred work must not retain cursor or result data until GC."""
+    store = _RecordingStore()
+    effect = _CoroutineReturningEffect()
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(),
+            effect,
+        )
+
+    assert caught.value.details == {"phase": "record_effect"}
+    assert effect.returned_coroutine.cr_frame is None
+    assert store.events == ["load"]
+
+
+def test_returned_future_is_cancelled_before_bounded_failure() -> None:
+    """Rejected pending asyncio futures must not remain live after the call."""
+    store = _RecordingStore()
+    effect = _FutureReturningEffect()
+    try:
+        with pytest.raises(ResultApplicationError) as caught:
+            apply_checkpointed_result_in_transaction(
+                object(),
+                store,
+                "result-writer",
+                _item(),
+                effect,
+            )
+
+        assert caught.value.details == {"phase": "record_effect"}
+        assert effect.returned_future.cancelled()
+        assert store.events == ["load"]
+    finally:
+        effect.close()
+
+
+def test_returned_concurrent_future_receives_cancellation_before_failure() -> None:
+    """Rejected concurrent futures must receive cancellation before returning."""
+    store = _RecordingStore()
+    effect = _ConcurrentFutureReturningEffect()
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(),
+            effect,
+        )
+
+    assert caught.value.details == {"phase": "record_effect"}
+    assert effect.returned_future.cancelled()
+    assert store.events == ["load"]
+
+
+def test_running_future_cannot_reuse_transaction_cursor_after_rejection() -> None:
+    """Already-running deferred work must lose package-supplied cursor authority."""
+    store = _RecordingStore()
+    cursor = _ExecutableCursor()
+    effect = _RunningConcurrentFutureReturningEffect()
+    try:
+        with pytest.raises(ResultApplicationError) as caught:
+            apply_checkpointed_result_in_transaction(
+                cursor,
+                store,
+                "result-writer",
+                _item(),
+                effect,
+            )
+
+        assert caught.value.details == {"phase": "record_effect"}
+        assert effect.returned_future is not None
+        effect.release.set()
+        with pytest.raises(ResultApplicationError) as worker_failure:
+            effect.returned_future.result(timeout=2)
+        assert worker_failure.value.details == {"phase": "record_effect"}
+        assert cursor.calls == []
+        assert store.events == ["load"]
+    finally:
+        effect.close()
diff --git a/tests/test_result_application_coverage_edges.py b/tests/test_result_application_coverage_edges.py
new file mode 100644
index 000000000..fb98a6e6d
--- /dev/null
+++ b/tests/test_result_application_coverage_edges.py
@@ -0,0 +1,274 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Coverage regressions for fail-closed transactional result application edges."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from pg_llm_batch.exceptions import ValidationError
+from pg_llm_batch.result_application import (
+    ResultApplicationError,
+    apply_checkpointed_result_in_transaction,
+)
+from pg_llm_batch.result_streaming import (
+    BatchResultCheckpoint,
+    CheckpointedBatchResultRecord,
+)
+
+
+class _IntegerSubclass(int):
+    """Represent a behavior-capable integer subtype accepted by checkpoint construction."""
+
+
+class _CheckpointSubclass(BatchResultCheckpoint):
+    """Represent a behavior-capable checkpoint subtype rejected at the apply boundary."""
+
+
+class _Store:
+    """Record checkpoint-store operations and return configured exact evidence."""
+
+    def __init__(
+        self,
+        *,
+        previous: BatchResultCheckpoint | None = None,
+        saved: BatchResultCheckpoint | None = None,
+    ) -> None:
+        self.previous = previous
+        self.saved = saved
+        self.events: list[str] = []
+
+    def load_in_transaction(self, *_args: object) -> BatchResultCheckpoint | None:
+        """Record one load and return the configured predecessor."""
+        self.events.append("load")
+        return self.previous
+
+    def save_in_transaction(
+        self,
+        _cursor: Any,
+        _consumer_name: str,
+        checkpoint: BatchResultCheckpoint,
+        *,
+        expected_previous: BatchResultCheckpoint | None = None,
+    ) -> BatchResultCheckpoint:
+        """Record one save and return the configured confirmation."""
+        self.events.append("save")
+        assert expected_previous is self.previous
+        return checkpoint if self.saved is None else self.saved
+
+
+def _checkpoint(
+    *,
+    checkpoint_type: type[BatchResultCheckpoint] = BatchResultCheckpoint,
+    schema_version: int = 1,
+    batch_id: str = "batch-123",
+    endpoint_alias: str = "openrouter",
+    file_kind: str = "result",
+    file_id: str = "file-123",
+    record_count: int = 1,
+    digest: str = "a" * 64,
+) -> BatchResultCheckpoint:
+    """Build one valid checkpoint while allowing one deliberate boundary variant."""
+    return checkpoint_type(
+        schema_version=schema_version,
+        batch_id=batch_id,
+        endpoint_alias=endpoint_alias,
+        file_kind=file_kind,
+        file_id=file_id,
+        file_line_number=record_count,
+        batch_line_count=record_count,
+        record_count=record_count,
+        prefix_sha256=digest,
+    )
+
+
+def _item(checkpoint: BatchResultCheckpoint) -> CheckpointedBatchResultRecord:
+    """Pair a checkpoint with one exact JSON-object result record."""
+    return CheckpointedBatchResultRecord(
+        batch_id="batch-123",
+        file_kind="result",
+        record={"custom_id": "request-1"},
+        checkpoint=checkpoint,
+    )
+
+
+def test_checkpoint_integer_subclass_is_rejected_before_store_access() -> None:
+    """Exact primitive checks must reject integer subclasses before persistence."""
+    checkpoint = _checkpoint(schema_version=_IntegerSubclass(1))
+    store = _Store()
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(checkpoint), lambda *_args: None
+        )
+
+    assert caught.value.details["field"] == "item.checkpoint.schema_version"
+    assert caught.value.details["value"] == "<redacted>"
+    assert store.events == []
+
+
+def test_checkpoint_subclass_is_rejected_before_store_access() -> None:
+    """The candidate checkpoint itself must be an exact package checkpoint type."""
+    checkpoint = _checkpoint(checkpoint_type=_CheckpointSubclass)
+    store = _Store()
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", _item(checkpoint), lambda *_args: None
+        )
+
+    assert caught.value.details["field"] == "item.checkpoint"
+    assert caught.value.details["value"] == "<redacted>"
+    assert store.events == []
+
+
+@pytest.mark.parametrize(
+    ("previous_batch_id", "previous_endpoint_alias"),
+    [
+        ("batch-other", "openrouter"),
+        ("batch-123", "secondary"),
+    ],
+)
+def test_loaded_checkpoint_identity_mismatch_fails_before_effect(
+    previous_batch_id: str,
+    previous_endpoint_alias: str,
+) -> None:
+    """Loaded exact checkpoints with a different identity must fail before effects."""
+    candidate = _checkpoint(record_count=2, digest="b" * 64)
+    previous = _checkpoint(
+        batch_id=previous_batch_id,
+        endpoint_alias=previous_endpoint_alias,
+        record_count=1,
+    )
+    store = _Store(previous=previous)
+    effects: list[str] = []
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda *_args: effects.append("effect"),
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_load"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load"]
+    assert effects == []
+
+
+@pytest.mark.parametrize(
+    ("previous_file_kind", "previous_file_id"),
+    [
+        ("error", "file-123"),
+        ("result", "file-other"),
+    ],
+)
+def test_loaded_checkpoint_file_identity_mismatch_fails_before_effect(
+    previous_file_kind: str,
+    previous_file_id: str,
+) -> None:
+    """A predecessor for another provider file must never authorize an effect."""
+    candidate = _checkpoint(record_count=2, digest="b" * 64)
+    previous = _checkpoint(
+        file_kind=previous_file_kind,
+        file_id=previous_file_id,
+        record_count=1,
+    )
+    store = _Store(previous=previous)
+    effects: list[str] = []
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda *_args: effects.append("effect"),
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_load"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load"]
+    assert effects == []
+
+
+def test_exact_mismatched_save_confirmation_is_rejected() -> None:
+    """An exact but different saved checkpoint must not forge application success."""
+    candidate = _checkpoint(record_count=1, digest="a" * 64)
+    saved = _checkpoint(record_count=2, digest="b" * 64)
+    store = _Store(saved=saved)
+    effects: list[str] = []
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda *_args: effects.append("effect"),
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_save"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load", "save"]
+    assert effects == ["effect"]
+
+
+def test_loaded_checkpoint_integer_subclass_is_rejected_before_effect() -> None:
+    """Loaded exact checkpoints must also enforce exact primitive evidence."""
+    candidate = _checkpoint(record_count=2, digest="b" * 64)
+    previous = _checkpoint(
+        schema_version=_IntegerSubclass(1),
+        record_count=1,
+        digest="a" * 64,
+    )
+    store = _Store(previous=previous)
+    effects: list[str] = []
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda *_args: effects.append("effect"),
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_load"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load"]
+    assert effects == []
+
+
+def test_saved_checkpoint_integer_subclass_is_rejected_after_effect() -> None:
+    """Save confirmation must enforce exact primitive evidence before success."""
+    candidate = _checkpoint(record_count=1, digest="a" * 64)
+    saved = _checkpoint(
+        schema_version=_IntegerSubclass(1),
+        record_count=1,
+        digest="a" * 64,
+    )
+    store = _Store(saved=saved)
+    effects: list[str] = []
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda *_args: effects.append("effect"),
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_save"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load", "save"]
+    assert effects == ["effect"]
diff --git a/tests/test_result_application_exact_type_boundary.py b/tests/test_result_application_exact_type_boundary.py
new file mode 100644
index 000000000..6972fa7a5
--- /dev/null
+++ b/tests/test_result_application_exact_type_boundary.py
@@ -0,0 +1,275 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Hostile-subclass regressions for transactional result application."""
+
+from __future__ import annotations
+
+import traceback
+from typing import Any
+
+import pytest
+
+from pg_llm_batch.exceptions import ValidationError
+from pg_llm_batch.result_application import (
+    ResultApplicationError,
+    apply_checkpointed_result_in_transaction,
+)
+from pg_llm_batch.result_streaming import (
+    BatchResultCheckpoint,
+    CheckpointedBatchResultRecord,
+)
+
+_SECRET_SENTINEL = "SECRET-SENTINEL hostile subclass diagnostic"
+
+
+def _checkpoint(*, record_count: int = 1, digest: str = "a" * 64) -> BatchResultCheckpoint:
+    """Build one valid checkpoint for exact-type boundary tests."""
+    return BatchResultCheckpoint(
+        schema_version=1,
+        batch_id="batch-123",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=record_count,
+        batch_line_count=record_count,
+        record_count=record_count,
+        prefix_sha256=digest,
+    )
+
+
+def _item(checkpoint: BatchResultCheckpoint) -> CheckpointedBatchResultRecord:
+    """Pair one ordinary JSON object with its checkpoint."""
+    return CheckpointedBatchResultRecord(
+        batch_id=checkpoint.batch_id,
+        file_kind=checkpoint.file_kind,
+        record={"custom_id": "request-1"},
+        checkpoint=checkpoint,
+    )
+
+
+class _RecordingStore:
+    """Record transaction-store access and return configured evidence."""
+
+    def __init__(
+        self,
+        *,
+        previous: BatchResultCheckpoint | None = None,
+        saved: BatchResultCheckpoint | None = None,
+    ) -> None:
+        self.previous = previous
+        self.saved = saved
+        self.events: list[str] = []
+
+    def load_in_transaction(self, *_args: object) -> BatchResultCheckpoint | None:
+        """Record and return the configured durable predecessor."""
+        self.events.append("load")
+        return self.previous
+
+    def save_in_transaction(
+        self,
+        _cursor: Any,
+        _consumer_name: str,
+        checkpoint: BatchResultCheckpoint,
+        *,
+        expected_previous: BatchResultCheckpoint | None = None,
+    ) -> BatchResultCheckpoint:
+        """Record and return configured save confirmation evidence."""
+        self.events.append("save")
+        assert expected_previous is self.previous
+        return checkpoint if self.saved is None else self.saved
+
+
+class _HostileItem(CheckpointedBatchResultRecord):
+    """Expose secret-bearing code if subclass attributes are trusted."""
+
+    def __getattribute__(self, name: str) -> Any:
+        """Raise before a caller can read the forged checkpoint attribute."""
+        if name == "checkpoint":
+            raise RuntimeError(_SECRET_SENTINEL)
+        return super().__getattribute__(name)
+
+
+class _HostileCheckpoint(BatchResultCheckpoint):
+    """Expose secret-bearing code if loaded checkpoint subclasses are trusted."""
+
+    def __post_init__(self) -> None:
+        """Keep fixture construction inert so product access is the first hostile read."""
+
+    def __getattribute__(self, name: str) -> Any:
+        """Raise before forged durable identity can be inspected."""
+        if name == "batch_id":
+            raise RuntimeError(_SECRET_SENTINEL)
+        return super().__getattribute__(name)
+
+
+class _AlwaysEqualCheckpoint(BatchResultCheckpoint):
+    """Forge equality so a mismatched save confirmation appears exact."""
+
+    def __eq__(self, _other: object) -> bool:
+        """Pretend every checkpoint equals this forged subclass instance."""
+        return True
+
+
+class _HostileIdentityText(str):
+    """Raise if an exact object trusts behavior-bearing string subclasses."""
+
+    def __ne__(self, _other: object) -> bool:
+        """Expose the sentinel if product code compares this forged identity."""
+        raise RuntimeError(_SECRET_SENTINEL)
+
+
+def _rendered_exception(error: BaseException) -> str:
+    """Render one traceback for confidentiality assertions."""
+    return "".join(traceback.format_exception(type(error), error, error.__traceback__))
+
+
+def test_hostile_item_subclass_is_rejected_before_attribute_access() -> None:
+    """Item validation must not execute caller-controlled subclass code."""
+    checkpoint = _checkpoint()
+    item = _HostileItem(
+        batch_id=checkpoint.batch_id,
+        file_kind=checkpoint.file_kind,
+        record={"custom_id": "request-1"},
+        checkpoint=checkpoint,
+    )
+    store = _RecordingStore()
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", item, lambda _cursor, _record: None
+        )
+
+    assert caught.value.details["field"] == "item"
+    assert caught.value.details["value"] == "<redacted>"
+    assert _SECRET_SENTINEL not in _rendered_exception(caught.value)
+    assert store.events == []
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("batch_id", _HostileIdentityText("batch-123")),
+        ("file_kind", _HostileIdentityText("result")),
+    ],
+)
+def test_hostile_item_identity_text_is_rejected_before_comparison(
+    field: str,
+    value: str,
+) -> None:
+    """Identity fields must be exact strings before equality can execute."""
+    checkpoint = _checkpoint()
+    item = CheckpointedBatchResultRecord(
+        batch_id=value if field == "batch_id" else checkpoint.batch_id,
+        file_kind=value if field == "file_kind" else checkpoint.file_kind,
+        record={"custom_id": "request-1"},
+        checkpoint=checkpoint,
+    )
+    store = _RecordingStore()
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", item, lambda _cursor, _record: None
+        )
+
+    assert caught.value.details["field"] == f"item.{field}"
+    assert caught.value.details["value"] == "<redacted>"
+    assert _SECRET_SENTINEL not in _rendered_exception(caught.value)
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == []
+
+
+def test_hostile_checkpoint_identity_text_is_rejected_before_comparison() -> None:
+    """Checkpoint identity fields must be exact strings before equality can execute."""
+    checkpoint = BatchResultCheckpoint(
+        schema_version=1,
+        batch_id=_HostileIdentityText("batch-123"),
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=1,
+        batch_line_count=1,
+        record_count=1,
+        prefix_sha256="a" * 64,
+    )
+    item = CheckpointedBatchResultRecord(
+        batch_id="batch-123",
+        file_kind="result",
+        record={"custom_id": "request-1"},
+        checkpoint=checkpoint,
+    )
+    store = _RecordingStore()
+
+    with pytest.raises(ValidationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(), store, "result-writer", item, lambda _cursor, _record: None
+        )
+
+    assert caught.value.details["field"] == "item.checkpoint.batch_id"
+    assert caught.value.details["value"] == "<redacted>"
+    assert _SECRET_SENTINEL not in _rendered_exception(caught.value)
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == []
+
+
+def test_hostile_loaded_checkpoint_subclass_is_rejected_before_identity_access() -> None:
+    """Loaded evidence validation must not execute forged checkpoint attributes."""
+    candidate = _checkpoint(record_count=2, digest="b" * 64)
+    previous = _HostileCheckpoint(
+        schema_version=1,
+        batch_id="batch-123",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=1,
+        batch_line_count=1,
+        record_count=1,
+        prefix_sha256="a" * 64,
+    )
+    store = _RecordingStore(previous=previous)
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda _cursor, _record: None,
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_load"}
+    assert _SECRET_SENTINEL not in _rendered_exception(caught.value)
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load"]
+
+
+def test_forged_save_confirmation_subclass_cannot_claim_exact_success() -> None:
+    """Save success must require an exact built-in checkpoint instance."""
+    candidate = _checkpoint(record_count=1, digest="a" * 64)
+    forged = _AlwaysEqualCheckpoint(
+        schema_version=1,
+        batch_id="different-batch",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="different-file",
+        file_line_number=2,
+        batch_line_count=2,
+        record_count=2,
+        prefix_sha256="b" * 64,
+    )
+    store = _RecordingStore(saved=forged)
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            _item(candidate),
+            lambda _cursor, _record: None,
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_save"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None
+    assert store.events == ["load", "save"]
diff --git a/tests/test_result_application_regression_guard.py b/tests/test_result_application_regression_guard.py
new file mode 100644
index 000000000..c33116ce7
--- /dev/null
+++ b/tests/test_result_application_regression_guard.py
@@ -0,0 +1,91 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Regression tests for pre-effect checkpoint monotonicity enforcement."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from pg_llm_batch.checkpoint_store import CheckpointConflictError
+from pg_llm_batch.result_application import apply_checkpointed_result_in_transaction
+from pg_llm_batch.result_streaming import BatchResultCheckpoint, CheckpointedBatchResultRecord
+
+
+def _checkpoint(*, record_count: int, digest: str) -> BatchResultCheckpoint:
+    """Build one valid checkpoint in a single batch-wide result stream."""
+    return BatchResultCheckpoint(
+        schema_version=1,
+        batch_id="batch-123",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=record_count,
+        batch_line_count=record_count,
+        record_count=record_count,
+        prefix_sha256=digest,
+    )
+
+
+class _RegressionStore:
+    """Expose a durable checkpoint ahead of the stale candidate under test."""
+
+    def __init__(self, previous: BatchResultCheckpoint) -> None:
+        self.previous = previous
+        self.events: list[str] = []
+
+    def load_in_transaction(
+        self,
+        _cursor: Any,
+        _consumer_name: str,
+        _batch_id: str,
+        _endpoint_alias: str,
+    ) -> BatchResultCheckpoint:
+        """Return the already advanced durable predecessor."""
+        self.events.append("load")
+        return self.previous
+
+    def save_in_transaction(
+        self,
+        _cursor: Any,
+        consumer_name: str,
+        checkpoint: BatchResultCheckpoint,
+        *,
+        expected_previous: BatchResultCheckpoint | None = None,
+    ) -> BatchResultCheckpoint:
+        """Model the built-in store's late checkpoint-regression rejection."""
+        self.events.append("save")
+        assert expected_previous == self.previous
+        raise CheckpointConflictError(
+            consumer_name,
+            checkpoint.batch_id,
+            "checkpoint_regression",
+        )
+
+
+def test_stale_checkpoint_regression_fails_before_caller_owned_effect() -> None:
+    """Detect an already-visible count regression before running business logic."""
+    previous = _checkpoint(record_count=2, digest="b" * 64)
+    candidate = _checkpoint(record_count=1, digest="a" * 64)
+    item = CheckpointedBatchResultRecord(
+        batch_id=candidate.batch_id,
+        file_kind=candidate.file_kind,
+        record={"custom_id": "request-1"},
+        checkpoint=candidate,
+    )
+    store = _RegressionStore(previous)
+
+    def effect(_cursor: Any, _record: dict[str, Any]) -> None:
+        store.events.append("effect")
+
+    with pytest.raises(CheckpointConflictError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            store,
+            "result-writer",
+            item,
+            effect,
+        )
+
+    assert caught.value.reason == "checkpoint_regression"
+    assert store.events == ["load"]
diff --git a/tests/test_result_application_save_confirmation.py b/tests/test_result_application_save_confirmation.py
new file mode 100644
index 000000000..c7ace4db4
--- /dev/null
+++ b/tests/test_result_application_save_confirmation.py
@@ -0,0 +1,86 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Regression test for durable checkpoint save confirmation."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from pg_llm_batch.result_application import (
+    ResultApplicationError,
+    apply_checkpointed_result_in_transaction,
+)
+from pg_llm_batch.result_streaming import (
+    BatchResultCheckpoint,
+    CheckpointedBatchResultRecord,
+)
+
+
+def _checkpoint(*, record_count: int, digest: str) -> BatchResultCheckpoint:
+    """Build one valid checkpoint with deterministic position evidence."""
+    return BatchResultCheckpoint(
+        schema_version=1,
+        batch_id="batch-123",
+        endpoint_alias="openrouter",
+        file_kind="result",
+        file_id="file-123",
+        file_line_number=record_count,
+        batch_line_count=record_count,
+        record_count=record_count,
+        prefix_sha256=digest,
+    )
+
+
+class _MismatchedSaveStore:
+    """Simulate a broken adapter that does not confirm the requested checkpoint."""
+
+    def __init__(self, returned_checkpoint: BatchResultCheckpoint) -> None:
+        self.returned_checkpoint = returned_checkpoint
+
+    def load_in_transaction(
+        self,
+        _cursor: Any,
+        _consumer_name: str,
+        _batch_id: str,
+        _endpoint_alias: str,
+    ) -> None:
+        """Report no durable predecessor for the fresh application."""
+        return None
+
+    def save_in_transaction(
+        self,
+        _cursor: Any,
+        _consumer_name: str,
+        _checkpoint: BatchResultCheckpoint,
+        *,
+        expected_previous: BatchResultCheckpoint | None = None,
+    ) -> BatchResultCheckpoint:
+        """Return a different durable identity to expose missing confirmation checks."""
+        assert expected_previous is None
+        return self.returned_checkpoint
+
+
+def test_mismatched_checkpoint_save_confirmation_fails_closed() -> None:
+    """Success must require the store to confirm the exact requested checkpoint."""
+    candidate = _checkpoint(record_count=1, digest="a" * 64)
+    mismatched = _checkpoint(record_count=2, digest="b" * 64)
+    item = CheckpointedBatchResultRecord(
+        batch_id=candidate.batch_id,
+        file_kind=candidate.file_kind,
+        record={"custom_id": "request-1"},
+        checkpoint=candidate,
+    )
+
+    with pytest.raises(ResultApplicationError) as caught:
+        apply_checkpointed_result_in_transaction(
+            object(),
+            _MismatchedSaveStore(mismatched),
+            "result-writer",
+            item,
+            lambda _cursor, _record: None,
+        )
+
+    assert caught.value.details == {"phase": "checkpoint_save"}
+    assert caught.value.__cause__ is None
+    assert caught.value.__context__ is None