refactor: 각 어댑터터별 리펙토링 진행

src/application-core/build.gradle

@@ -1,6 +1,5 @@
 // Framework-free application use-case contract. Runtime dependencies are project-only;
 // composition and diagnostic rendering belong to adapters/bootstrap.
-apply from: "${rootProject.projectDir}/gradle/strict-qualification-test.gradle"
 
 dependencies {
     implementation project(':shared-contract')

src/application-core/src/main/java/dev/caskeleton/application/fileserver/api/metadata/UploadSession.java

@@ -14,6 +14,10 @@ import java.util.UUID;
  *
  * <p>{@code committedOffset} advances only by bytes proven durable, and the lease columns make the
  * single-writer rule enforceable across instances.
+ *
+ * <p>{@code terminal} says whether anyone may still write to this upload at all, which the lease
+ * cannot: a lease answers "who is writing now", and an upload that has been cancelled or has failed
+ * verification simply has no lease — indistinguishable, until this flag, from an idle live one.
  */
 public record UploadSession(
     UploadId uploadId,
@@ -25,6 +29,7 @@ public record UploadSession(
     Optional<String> leaseOwner,
     Optional<UUID> leaseToken,
     Optional<Instant> leaseUntil,
+    boolean terminal,
     long version,
     Instant createdAt,
     Instant updatedAt) {

src/application-core/src/main/java/dev/caskeleton/application/fileserver/api/metadata/UploadSessionStore.java

@@ -39,5 +39,28 @@ public interface UploadSessionStore {
 
   void releaseLease(UploadId uploadId, WriterLease lease);
 
+  /**
+   * Ends the upload's writable life.
+   *
+   * <p>Called from the transaction that decides the upload is over and queues its staging cleanup,
+   * so "nobody may write here again" and "these bytes are scheduled for deletion" become true
+   * together. Queuing the cleanup on its own left a window in which a writer could still take a
+   * lease on the object about to be removed.
+   *
+   * @return whether this call is the one that ended it
+   */
+  boolean terminate(UploadId uploadId);
+
+  /**
+   * Claims a terminal upload whose writer lease has lapsed, so its staged bytes can be deleted.
+   *
+   * <p>A claim, not a question. Reading the lease and then deleting leaves room for a writer to
+   * acquire that lease in between, and the object removed is then one an upload is actively
+   * appending to. The store settles it in the database and refuses writers afterwards.
+   *
+   * @return whether the caller may delete the staged bytes
+   */
+  boolean claimForCleanup(UploadId uploadId, Instant now);
+
   List<UploadSession> findExpired(Instant cutoff, int limit);
 }

src/application-core/src/main/java/dev/caskeleton/application/fileserver/cleanup/CleanupQueue.java

@@ -19,4 +19,17 @@ public interface CleanupQueue {
   void markDone(CleanupItem item);
 
   void markFailed(CleanupItem item, String reasonCode, Instant nextAttemptAt);
+
+  /**
+   * Returns items whose worker died mid-claim to the queue.
+   *
+   * <p>A claim moves an item out of the due set, so an item whose worker performed the physical
+   * delete and then died is invisible to {@link #claimDue}: nothing reclaims it, and the file's
+   * quota and lifecycle stay unsettled for good. This is the only path back.
+   *
+   * @param now the moment expiry is judged against
+   * @param limit how many items one pass may reclaim
+   * @return how many items came back
+   */
+  int reclaimExpiredClaims(Instant now, int limit);
 }

src/application-core/src/main/java/dev/caskeleton/application/fileserver/cleanup/DefaultCleanupService.java

@@ -10,7 +10,6 @@ import dev.caskeleton.application.fileserver.api.metadata.FileMetadataStore;
 import dev.caskeleton.application.fileserver.api.metadata.FileRecord;
 import dev.caskeleton.application.fileserver.api.metadata.FileRecordMutation;
 import dev.caskeleton.application.fileserver.api.metadata.QuotaScope;
-import dev.caskeleton.application.fileserver.api.metadata.UploadSession;
 import dev.caskeleton.application.fileserver.api.metadata.UploadSessionStore;
 import dev.caskeleton.application.fileserver.observability.FileserverMetricsPort;
 import dev.caskeleton.application.transaction.TransactionPort;
@@ -68,6 +67,11 @@ public final class DefaultCleanupService implements CleanupService {
       throw new IllegalArgumentException("batch bounds must be positive");
     }
     Instant now = clock.instant();
+    // Abandoned claims come back first. A claim takes an item out of the due set, so an item whose
+    // worker performed the physical delete and then died is invisible to claimDue — nothing ever
+    // reclaimed it, and the file's quota and lifecycle stayed unsettled for good. Reclaiming before
+    // claiming means the recovered items are eligible in this same pass.
+    transactions.inWrite(() -> queue.reclaimExpiredClaims(now, maxItems));
     List<CleanupItem> due = transactions.inWrite(() -> queue.claimDue(now, maxItems));
 
     int deleted = 0;
@@ -128,8 +132,11 @@ public final class DefaultCleanupService implements CleanupService {
    * upload that is mid-flight, so the item is deferred rather than executed.
    */
   private Outcome cleanStaging(CleanupItem item, UploadId uploadId, Instant now) {
-    Optional<UploadSession> session = sessionStore.find(uploadId);
-    if (session.isPresent() && isLeaseActive(session.get(), now)) {
+    // Claimed in the database rather than decided from a lease read a moment ago. Between the read
+    // and the delete a writer could acquire that very lease, and the bytes removed were then the
+    // ones an upload was actively appending to. The claim clears the lease in the same statement
+    // that proves it was not held, and the session is terminal, so no later acquire can succeed.
+    if (sessionStore.find(uploadId).isPresent() && !sessionStore.claimForCleanup(uploadId, now)) {
       markFailed(item, "ACTIVE_WRITER_LEASE", now);
       return Outcome.of(OutcomeKind.SKIPPED_ACTIVE_LEASE, 0);
     }
@@ -215,10 +222,6 @@ public final class DefaultCleanupService implements CleanupService {
         || record.state() == FileState.EXPIRED;
   }
 
-  private static boolean isLeaseActive(UploadSession session, Instant now) {
-    return session.leaseUntil().map(until -> until.isAfter(now)).orElse(false);
-  }
-
   /** Per-item outcome, kept internal so the batch result stays the only public shape. */
   private record Outcome(OutcomeKind kind, long reclaimedBytes) {
 

src/application-core/src/main/java/dev/caskeleton/application/fileserver/upload/DefaultFinalizeUploadService.java

@@ -243,6 +243,10 @@ public final class DefaultFinalizeUploadService implements FinalizeUploadService
                       FileState.VERIFYING,
                       target,
                       FileRecordMutation.failure(verdict.code()));
+              // Terminal alongside the queued cleanup: a rejected upload is finished being written
+              // to, and leaving it ACTIVE let a writer acquire a lease on bytes already scheduled
+              // for deletion.
+              sessionStore.terminate(session.uploadId());
               cleanupQueue.enqueue(
                   CleanupRequest.forStaging(
                       CleanupType.FAILED_VERIFICATION_CONTENT, moved.fileId(), session.uploadId()));

src/application-core/src/main/java/dev/caskeleton/application/fileserver/upload/DefaultUploadApplicationService.java

@@ -253,6 +253,10 @@ public final class DefaultUploadApplicationService implements UploadApplicationS
           // scheduled. Both writes commit together, so no cancel can leave a file unreachable with
           // nothing queued to reclaim it.
           metadataStore.markDeleting(record.fileId(), record.version());
+          // Terminal in the same transaction that queues the cleanup. Queuing alone left the
+          // session ACTIVE, so a writer could still take a lease on the very bytes the cleanup was
+          // about to delete and the two raced for the same object.
+          sessionStore.terminate(session.uploadId());
           cleanupQueue.enqueue(
               CleanupRequest.forStaging(
                   CleanupType.CANCELLED_STAGING, record.fileId(), session.uploadId()));

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyClaimAttempt.java

@@ -1,15 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-/** Caller-retained owner and operation tokens allocated before the first provider send. */
-public record IdempotencyClaimAttempt(String ownerToken, String operationId) {
-
-  public IdempotencyClaimAttempt {
-    ownerToken = IdempotencyV2Validation.opaqueToken(ownerToken, "ownerToken");
-    operationId = IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyClaimAttempt[REDACTED]";
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyClaimOutcome.java

@@ -1,85 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.util.Objects;
-
-/** Typed result of the atomic request-replay claim state machine. */
-public sealed interface IdempotencyClaimOutcome {
-
-  record Acquired(IdempotencyOwner owner, Instant processingLeaseUntil)
-      implements IdempotencyClaimOutcome {
-
-    public Acquired {
-      Objects.requireNonNull(owner, "owner must be non-null");
-      IdempotencyV2Validation.instant(processingLeaseUntil, "processingLeaseUntil");
-    }
-  }
-
-  record ReplayedAcquire(IdempotencyOwner owner, Instant processingLeaseUntil)
-      implements IdempotencyClaimOutcome {
-
-    public ReplayedAcquire {
-      Objects.requireNonNull(owner, "owner must be non-null");
-      IdempotencyV2Validation.instant(processingLeaseUntil, "processingLeaseUntil");
-    }
-  }
-
-  record TakenOverClaimed(IdempotencyOwner owner, Instant processingLeaseUntil)
-      implements IdempotencyClaimOutcome {
-
-    public TakenOverClaimed {
-      Objects.requireNonNull(owner, "owner must be non-null");
-      IdempotencyV2Validation.instant(processingLeaseUntil, "processingLeaseUntil");
-    }
-  }
-
-  record CompletedReplay(StoredResponse response, Instant replayUntil)
-      implements IdempotencyClaimOutcome {
-
-    public CompletedReplay {
-      Objects.requireNonNull(response, "response must be non-null");
-      IdempotencyV2Validation.instant(replayUntil, "replayUntil");
-    }
-
-    @Override
-    public String toString() {
-      return "CompletedReplay[response=REDACTED, replayUntil=" + replayUntil + "]";
-    }
-  }
-
-  record InProgress(Duration retryAfter, long currentAttempt) implements IdempotencyClaimOutcome {
-
-    public InProgress {
-      retryAfter =
-          IdempotencyV2Validation.positiveBounded(
-              retryAfter, IdempotencyV2Validation.MAXIMUM_RETRY_AFTER, "retryAfter");
-      currentAttempt = IdempotencyV2Validation.positiveAttempt(currentAttempt, "currentAttempt");
-    }
-  }
-
-  record RecoveryRequired(long currentAttempt) implements IdempotencyClaimOutcome {
-
-    public RecoveryRequired {
-      currentAttempt = IdempotencyV2Validation.positiveAttempt(currentAttempt, "currentAttempt");
-    }
-  }
-
-  record FingerprintMismatch() implements IdempotencyClaimOutcome {}
-
-  record OwnerOperationConflict() implements IdempotencyClaimOutcome {}
-
-  record Indeterminate(String operationId) implements IdempotencyClaimOutcome {
-
-    public Indeterminate {
-      operationId = IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-    }
-
-    @Override
-    public String toString() {
-      return "Indeterminate[operationId=REDACTED]";
-    }
-  }
-
-  record Unavailable() implements IdempotencyClaimOutcome {}
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyClaimRequest.java

@@ -1,60 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.time.Duration;
-import java.util.Objects;
-
-/**
- * Atomic claim inputs with separate processing lease and durable recovery/replay retention.
- *
- * <p>{@code replayTtl} retains in-progress execution evidence as well as a completed response. It
- * must outlive the processing lease so an expired {@code EXECUTING} record becomes recovery
- * required instead of disappearing and being unsafely re-executed.
- */
-public record IdempotencyClaimRequest(
-    IdempotencyScope scope,
-    RequestFingerprint fingerprint,
-    IdempotencyClaimAttempt claimAttempt,
-    Duration processingLeaseTtl,
-    Duration replayTtl,
-    String responseCodecId,
-    String policyRevision) {
-
-  public IdempotencyClaimRequest {
-    Objects.requireNonNull(scope, "scope must be non-null");
-    Objects.requireNonNull(fingerprint, "fingerprint must be non-null");
-    Objects.requireNonNull(claimAttempt, "claimAttempt must be non-null");
-    processingLeaseTtl =
-        IdempotencyV2Validation.positiveBounded(
-            processingLeaseTtl,
-            IdempotencyV2Validation.MAXIMUM_PROCESSING_LEASE,
-            "processingLeaseTtl");
-    replayTtl =
-        IdempotencyV2Validation.positiveBounded(
-            replayTtl, IdempotencyV2Validation.MAXIMUM_REPLAY_TTL, "replayTtl");
-    if (replayTtl.compareTo(processingLeaseTtl) <= 0) {
-      throw new IllegalArgumentException(
-          "replayTtl recovery retention must outlive processingLeaseTtl");
-    }
-    responseCodecId = IdempotencyV2Validation.boundedId(responseCodecId, "responseCodecId");
-    policyRevision = IdempotencyV2Validation.boundedId(policyRevision, "policyRevision");
-  }
-
-  /** Retention used for in-progress recovery evidence before the record becomes replayable. */
-  public Duration recoveryRetention() {
-    return replayTtl;
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyClaimRequest[scope=REDACTED, fingerprint=REDACTED, "
-        + "claimAttempt=REDACTED, processingLeaseTtl="
-        + processingLeaseTtl
-        + ", replayTtl="
-        + replayTtl
-        + ", responseCodecId="
-        + responseCodecId
-        + ", policyRevision="
-        + policyRevision
-        + "]";
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyCompleteOutcome.java

@@ -1,55 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Owner-safe response completion result with same-result replay and conflict separation. */
-public record IdempotencyCompleteOutcome(Status status, String operationId) {
-
-  public IdempotencyCompleteOutcome {
-    Objects.requireNonNull(status, "status must be non-null");
-    operationId = validateOperation(status, operationId);
-  }
-
-  public static IdempotencyCompleteOutcome responseConflict() {
-    return new IdempotencyCompleteOutcome(Status.RESPONSE_CONFLICT, null);
-  }
-
-  public static IdempotencyCompleteOutcome operationConflict() {
-    return new IdempotencyCompleteOutcome(Status.OPERATION_CONFLICT, null);
-  }
-
-  public static IdempotencyCompleteOutcome indeterminate(String operationId) {
-    return new IdempotencyCompleteOutcome(Status.INDETERMINATE, operationId);
-  }
-
-  public static IdempotencyCompleteOutcome unavailable() {
-    return new IdempotencyCompleteOutcome(Status.UNAVAILABLE, null);
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyCompleteOutcome[status=" + status + ", operationId=REDACTED]";
-  }
-
-  private static String validateOperation(Status status, String operationId) {
-    if (status == Status.INDETERMINATE) {
-      return IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-    }
-    if (operationId != null) {
-      throw new IllegalArgumentException("operationId is valid only for INDETERMINATE");
-    }
-    return null;
-  }
-
-  public enum Status {
-    COMPLETED,
-    ALREADY_COMPLETED_SAME_RESULT,
-    RESPONSE_CONFLICT,
-    ABSENT,
-    NOT_OWNER,
-    NOT_IN_PROGRESS,
-    OPERATION_CONFLICT,
-    INDETERMINATE,
-    UNAVAILABLE
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyExecutorV2.java

@@ -1,230 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.time.Duration;
-import java.util.Objects;
-
-/**
- * Owner-safe request-replay lifecycle.
- *
- * <p>The action runs only after a confirmed {@code STARTED}. This orchestration preserves
- * request-replay evidence but does not create a cross-store exactly-once boundary.
- */
-public final class IdempotencyExecutorV2 {
-
-  private final IdempotencyStorePortV2 store;
-  private final Duration processingLeaseTtl;
-  private final Duration replayTtl;
-  private final Duration failureRetention;
-  private final String responseCodecId;
-  private final String policyRevision;
-
-  public IdempotencyExecutorV2(
-      IdempotencyStorePortV2 store,
-      Duration processingLeaseTtl,
-      Duration replayTtl,
-      Duration failureRetention,
-      String responseCodecId,
-      String policyRevision) {
-    this.store = Objects.requireNonNull(store, "store must be non-null");
-    this.processingLeaseTtl = Objects.requireNonNull(processingLeaseTtl);
-    this.replayTtl = Objects.requireNonNull(replayTtl);
-    this.failureRetention = Objects.requireNonNull(failureRetention);
-    this.responseCodecId = Objects.requireNonNull(responseCodecId);
-    this.policyRevision = Objects.requireNonNull(policyRevision);
-    new IdempotencyClaimRequest(
-        IdempotencyScope.of("validation", "validation", "validation"),
-        new RequestFingerprint("0".repeat(64)),
-        new IdempotencyClaimAttempt("validation_owner", "validation_operation"),
-        processingLeaseTtl,
-        replayTtl,
-        responseCodecId,
-        policyRevision);
-    IdempotencyV2Validation.positiveBounded(
-        failureRetention, IdempotencyV2Validation.MAXIMUM_REPLAY_TTL, "failureRetention");
-  }
-
-  public IdempotencyClaimAttempt newAttempt(String operationId) {
-    return store.newClaimAttempt(operationId);
-  }
-
-  public <R> R execute(
-      IdempotencyScope scope,
-      RequestFingerprint fingerprint,
-      IdempotencyClaimAttempt attempt,
-      IdempotentAction<R> action,
-      IdempotentResponseCodec<R> codec) {
-    Objects.requireNonNull(action, "action must be non-null");
-    Objects.requireNonNull(codec, "codec must be non-null");
-    IdempotencyClaimRequest request =
-        new IdempotencyClaimRequest(
-            scope,
-            fingerprint,
-            attempt,
-            processingLeaseTtl,
-            replayTtl,
-            responseCodecId,
-            policyRevision);
-    IdempotencyClaimOutcome claim = store.claim(request);
-    if (claim instanceof IdempotencyClaimOutcome.CompletedReplay replay) {
-      return codec.deserialize(replay.response().payload());
-    }
-    if (claim instanceof IdempotencyClaimOutcome.FingerprintMismatch) {
-      throw new IdempotencyRequestMismatchException(scope);
-    }
-    if (claim instanceof IdempotencyClaimOutcome.InProgress) {
-      throw new IdempotencyInFlightException(scope);
-    }
-    if (claim instanceof IdempotencyClaimOutcome.RecoveryRequired
-        || claim instanceof IdempotencyClaimOutcome.OwnerOperationConflict) {
-      throw recovery("claim requires reconciliation");
-    }
-    if (claim instanceof IdempotencyClaimOutcome.Unavailable) {
-      throw new IdempotencyUnavailableException();
-    }
-    if (claim instanceof IdempotencyClaimOutcome.Indeterminate) {
-      return reconcileClaim(request, action, codec);
-    }
-    IdempotencyOwner owner =
-        switch (claim) {
-          case IdempotencyClaimOutcome.Acquired acquired -> acquired.owner();
-          case IdempotencyClaimOutcome.ReplayedAcquire replayed -> replayed.owner();
-          case IdempotencyClaimOutcome.TakenOverClaimed takenOver -> takenOver.owner();
-          default -> throw recovery("unsupported claim outcome");
-        };
-    return startAndRun(request, owner, action, codec, false);
-  }
-
-  private <R> R reconcileClaim(
-      IdempotencyClaimRequest request,
-      IdempotentAction<R> action,
-      IdempotentResponseCodec<R> codec) {
-    IdempotencyInspection inspection =
-        store.inspect(
-            new IdempotencyInspectionRequest(
-                request.scope(), request.fingerprint(), request.claimAttempt()));
-    return switch (inspection) {
-      case IdempotencyInspection.ClaimedSameOperation claimed ->
-          startAndRun(request, claimed.owner(), action, codec, false);
-      case IdempotencyInspection.ExecutingSameOperation executing ->
-          runStarted(request, executing.owner(), action, codec);
-      case IdempotencyInspection.CompletedReplay replay ->
-          codec.deserialize(replay.response().payload());
-      case IdempotencyInspection.FingerprintMismatch ignored ->
-          throw new IdempotencyRequestMismatchException(request.scope());
-      case IdempotencyInspection.Unavailable ignored -> throw new IdempotencyUnavailableException();
-      default -> throw recovery("indeterminate claim cannot be safely resumed");
-    };
-  }
-
-  private <R> R startAndRun(
-      IdempotencyClaimRequest request,
-      IdempotencyOwner owner,
-      IdempotentAction<R> action,
-      IdempotentResponseCodec<R> codec,
-      boolean retriedStart) {
-    String operationId = request.claimAttempt().operationId();
-    IdempotencyStartOutcome started = store.markExecutionStarted(owner, operationId);
-    if (started.status() == IdempotencyStartOutcome.Status.INDETERMINATE && !retriedStart) {
-      IdempotencyInspection inspection =
-          store.inspect(
-              new IdempotencyInspectionRequest(
-                  request.scope(), request.fingerprint(), request.claimAttempt()));
-      if (inspection instanceof IdempotencyInspection.ClaimedSameOperation claimed) {
-        return startAndRun(request, claimed.owner(), action, codec, true);
-      }
-      if (inspection instanceof IdempotencyInspection.ExecutingSameOperation executing) {
-        return runStarted(request, executing.owner(), action, codec);
-      }
-      if (inspection instanceof IdempotencyInspection.CompletedReplay replay) {
-        return codec.deserialize(replay.response().payload());
-      }
-      throw recovery("execution start is indeterminate");
-    }
-    if (started.status() == IdempotencyStartOutcome.Status.UNAVAILABLE) {
-      throw new IdempotencyUnavailableException();
-    }
-    if (started.status() != IdempotencyStartOutcome.Status.STARTED
-        && started.status() != IdempotencyStartOutcome.Status.ALREADY_STARTED_SAME_OPERATION) {
-      throw recovery("execution start was not confirmed for the exact operation");
-    }
-    return runStarted(request, owner, action, codec);
-  }
-
-  private <R> R runStarted(
-      IdempotencyClaimRequest request,
-      IdempotencyOwner owner,
-      IdempotentAction<R> action,
-      IdempotentResponseCodec<R> codec) {
-    String operationId = request.claimAttempt().operationId();
-    IdempotentAction.Outcome<R> outcome;
-    try {
-      outcome = Objects.requireNonNull(action.run(), "action outcome must be non-null");
-    } catch (RuntimeException failure) {
-      preserveUnknown(owner, operationId);
-      throw failure;
-    }
-    return switch (outcome) {
-      case IdempotentAction.Outcome.Success<R> success ->
-          complete(request, owner, operationId, success.result(), codec, false);
-      case IdempotentAction.Outcome.RetryableNoEffect<R> retryable -> {
-        store.markFailed(
-            owner,
-            IdempotencyFailureDisposition.RETRYABLE_NO_EFFECT,
-            failureRetention,
-            operationId);
-        throw retryable.failure();
-      }
-      case IdempotentAction.Outcome.EffectUnknown<R> unknown -> {
-        preserveUnknown(owner, operationId);
-        throw unknown.failure();
-      }
-    };
-  }
-
-  private <R> R complete(
-      IdempotencyClaimRequest request,
-      IdempotencyOwner owner,
-      String operationId,
-      R result,
-      IdempotentResponseCodec<R> codec,
-      boolean retried) {
-    StoredResponse response = new StoredResponse(codec.serialize(result));
-    IdempotencyCompleteOutcome completed = store.complete(owner, response, replayTtl, operationId);
-    if (completed.status() == IdempotencyCompleteOutcome.Status.COMPLETED
-        || completed.status() == IdempotencyCompleteOutcome.Status.ALREADY_COMPLETED_SAME_RESULT) {
-      return result;
-    }
-    if (completed.status() == IdempotencyCompleteOutcome.Status.INDETERMINATE) {
-      IdempotencyInspection inspection =
-          store.inspect(
-              new IdempotencyInspectionRequest(
-                  request.scope(), request.fingerprint(), request.claimAttempt()));
-      if (inspection instanceof IdempotencyInspection.CompletedReplay replay) {
-        if (response.equals(replay.response())) {
-          return result;
-        }
-        throw recovery("completion replay conflicts with the local response");
-      }
-      if (inspection instanceof IdempotencyInspection.ExecutingSameOperation && !retried) {
-        return complete(request, owner, operationId, result, codec, true);
-      }
-      throw recovery("completion response is indeterminate and could not be reconciled");
-    }
-    if (completed.status() == IdempotencyCompleteOutcome.Status.UNAVAILABLE) {
-      throw new IdempotencyUnavailableException();
-    }
-    throw recovery("completion was not confirmed");
-  }
-
-  private void preserveUnknown(IdempotencyOwner owner, String operationId) {
-    store.markFailed(
-        owner,
-        IdempotencyFailureDisposition.ABANDONED_EFFECT_UNKNOWN,
-        failureRetention,
-        operationId);
-  }
-
-  private static IdempotencyRecoveryRequiredException recovery(String message) {
-    return new IdempotencyRecoveryRequiredException(message);
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyFailOutcome.java

@@ -1,51 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Owner-safe failed/abandoned transition result. */
-public record IdempotencyFailOutcome(Status status, String operationId) {
-
-  public IdempotencyFailOutcome {
-    Objects.requireNonNull(status, "status must be non-null");
-    operationId = validateOperation(status, operationId);
-  }
-
-  public static IdempotencyFailOutcome operationConflict() {
-    return new IdempotencyFailOutcome(Status.OPERATION_CONFLICT, null);
-  }
-
-  public static IdempotencyFailOutcome indeterminate(String operationId) {
-    return new IdempotencyFailOutcome(Status.INDETERMINATE, operationId);
-  }
-
-  public static IdempotencyFailOutcome unavailable() {
-    return new IdempotencyFailOutcome(Status.UNAVAILABLE, null);
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyFailOutcome[status=" + status + ", operationId=REDACTED]";
-  }
-
-  private static String validateOperation(Status status, String operationId) {
-    if (status == Status.INDETERMINATE) {
-      return IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-    }
-    if (operationId != null) {
-      throw new IllegalArgumentException("operationId is valid only for INDETERMINATE");
-    }
-    return null;
-  }
-
-  public enum Status {
-    MARKED_RETRYABLE,
-    MARKED_ABANDONED,
-    ALREADY_MARKED_SAME_OPERATION,
-    ABSENT,
-    NOT_OWNER,
-    NOT_IN_PROGRESS,
-    OPERATION_CONFLICT,
-    INDETERMINATE,
-    UNAVAILABLE
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyFailureDisposition.java

@@ -1,7 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-/** Application-confirmed effect disposition after execution started. */
-public enum IdempotencyFailureDisposition {
-  RETRYABLE_NO_EFFECT,
-  ABANDONED_EFFECT_UNKNOWN
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyInspection.java

@@ -1,69 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.time.Instant;
-import java.util.Objects;
-
-/** Read-only result used to reconcile claim/start responses without creating a new owner. */
-public sealed interface IdempotencyInspection {
-
-  record Absent() implements IdempotencyInspection {}
-
-  record ClaimedSameOperation(IdempotencyOwner owner, Instant processingLeaseUntil)
-      implements IdempotencyInspection {
-
-    public ClaimedSameOperation {
-      Objects.requireNonNull(owner, "owner must be non-null");
-      IdempotencyV2Validation.instant(processingLeaseUntil, "processingLeaseUntil");
-    }
-  }
-
-  record ExecutingSameOperation(IdempotencyOwner owner, Instant processingLeaseUntil)
-      implements IdempotencyInspection {
-
-    public ExecutingSameOperation {
-      Objects.requireNonNull(owner, "owner must be non-null");
-      IdempotencyV2Validation.instant(processingLeaseUntil, "processingLeaseUntil");
-    }
-  }
-
-  record CompletedReplay(StoredResponse response, Instant replayUntil)
-      implements IdempotencyInspection {
-
-    public CompletedReplay {
-      Objects.requireNonNull(response, "response must be non-null");
-      IdempotencyV2Validation.instant(replayUntil, "replayUntil");
-    }
-
-    @Override
-    public String toString() {
-      return "CompletedReplay[response=REDACTED, replayUntil=" + replayUntil + "]";
-    }
-  }
-
-  record InProgressOther(long currentAttempt) implements IdempotencyInspection {
-
-    public InProgressOther {
-      currentAttempt = IdempotencyV2Validation.positiveAttempt(currentAttempt, "currentAttempt");
-    }
-  }
-
-  record FailedRetryable(long currentAttempt) implements IdempotencyInspection {
-
-    public FailedRetryable {
-      currentAttempt = IdempotencyV2Validation.positiveAttempt(currentAttempt, "currentAttempt");
-    }
-  }
-
-  record Abandoned(long currentAttempt) implements IdempotencyInspection {
-
-    public Abandoned {
-      currentAttempt = IdempotencyV2Validation.positiveAttempt(currentAttempt, "currentAttempt");
-    }
-  }
-
-  record FingerprintMismatch() implements IdempotencyInspection {}
-
-  record OperationConflict() implements IdempotencyInspection {}
-
-  record Unavailable() implements IdempotencyInspection {}
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyInspectionRequest.java

@@ -1,19 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Read-only reconciliation input for a retained claim attempt after an uncertain response. */
-public record IdempotencyInspectionRequest(
-    IdempotencyScope scope, RequestFingerprint fingerprint, IdempotencyClaimAttempt claimAttempt) {
-
-  public IdempotencyInspectionRequest {
-    Objects.requireNonNull(scope, "scope must be non-null");
-    Objects.requireNonNull(fingerprint, "fingerprint must be non-null");
-    Objects.requireNonNull(claimAttempt, "claimAttempt must be non-null");
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyInspectionRequest[REDACTED]";
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyOwner.java

@@ -1,18 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Owner-safe handle returned by a successful v2 claim. */
-public record IdempotencyOwner(IdempotencyScope scope, String ownerToken, long attempt) {
-
-  public IdempotencyOwner {
-    Objects.requireNonNull(scope, "scope must be non-null");
-    ownerToken = IdempotencyV2Validation.opaqueToken(ownerToken, "ownerToken");
-    attempt = IdempotencyV2Validation.positiveAttempt(attempt, "attempt");
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyOwner[attempt=" + attempt + ", scope=REDACTED, ownerToken=REDACTED]";
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyReleaseOutcome.java

@@ -1,54 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Owner-safe release result valid only before execution starts. */
-public record IdempotencyReleaseOutcome(Status status, String operationId) {
-
-  public IdempotencyReleaseOutcome {
-    Objects.requireNonNull(status, "status must be non-null");
-    operationId = validateOperation(status, operationId);
-  }
-
-  public static IdempotencyReleaseOutcome executionAlreadyStarted() {
-    return new IdempotencyReleaseOutcome(Status.EXECUTION_ALREADY_STARTED, null);
-  }
-
-  public static IdempotencyReleaseOutcome operationConflict() {
-    return new IdempotencyReleaseOutcome(Status.OPERATION_CONFLICT, null);
-  }
-
-  public static IdempotencyReleaseOutcome indeterminate(String operationId) {
-    return new IdempotencyReleaseOutcome(Status.INDETERMINATE, operationId);
-  }
-
-  public static IdempotencyReleaseOutcome unavailable() {
-    return new IdempotencyReleaseOutcome(Status.UNAVAILABLE, null);
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyReleaseOutcome[status=" + status + ", operationId=REDACTED]";
-  }
-
-  private static String validateOperation(Status status, String operationId) {
-    if (status == Status.INDETERMINATE) {
-      return IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-    }
-    if (operationId != null) {
-      throw new IllegalArgumentException("operationId is valid only for INDETERMINATE");
-    }
-    return null;
-  }
-
-  public enum Status {
-    RELEASED_BEFORE_EXECUTION,
-    ALREADY_RELEASED_SAME_OPERATION,
-    ABSENT,
-    NOT_OWNER,
-    EXECUTION_ALREADY_STARTED,
-    OPERATION_CONFLICT,
-    INDETERMINATE,
-    UNAVAILABLE
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyRenewOutcome.java

@@ -1,50 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Owner-safe processing lease renewal result. */
-public record IdempotencyRenewOutcome(Status status, String operationId) {
-
-  public IdempotencyRenewOutcome {
-    Objects.requireNonNull(status, "status must be non-null");
-    operationId = validateOperation(status, operationId);
-  }
-
-  public static IdempotencyRenewOutcome operationConflict() {
-    return new IdempotencyRenewOutcome(Status.OPERATION_CONFLICT, null);
-  }
-
-  public static IdempotencyRenewOutcome indeterminate(String operationId) {
-    return new IdempotencyRenewOutcome(Status.INDETERMINATE, operationId);
-  }
-
-  public static IdempotencyRenewOutcome unavailable() {
-    return new IdempotencyRenewOutcome(Status.UNAVAILABLE, null);
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyRenewOutcome[status=" + status + ", operationId=REDACTED]";
-  }
-
-  private static String validateOperation(Status status, String operationId) {
-    if (status == Status.INDETERMINATE) {
-      return IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-    }
-    if (operationId != null) {
-      throw new IllegalArgumentException("operationId is valid only for INDETERMINATE");
-    }
-    return null;
-  }
-
-  public enum Status {
-    RENEWED,
-    ALREADY_RENEWED_SAME_OPERATION,
-    ABSENT,
-    NOT_OWNER,
-    NOT_IN_PROGRESS,
-    OPERATION_CONFLICT,
-    INDETERMINATE,
-    UNAVAILABLE
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyStartOutcome.java

@@ -1,50 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.util.Objects;
-
-/** Owner-safe {@code CLAIMED -> EXECUTING} transition result. */
-public record IdempotencyStartOutcome(Status status, String operationId) {
-
-  public IdempotencyStartOutcome {
-    Objects.requireNonNull(status, "status must be non-null");
-    operationId = validateOperation(status, operationId);
-  }
-
-  public static IdempotencyStartOutcome operationConflict() {
-    return new IdempotencyStartOutcome(Status.OPERATION_CONFLICT, null);
-  }
-
-  public static IdempotencyStartOutcome indeterminate(String operationId) {
-    return new IdempotencyStartOutcome(Status.INDETERMINATE, operationId);
-  }
-
-  public static IdempotencyStartOutcome unavailable() {
-    return new IdempotencyStartOutcome(Status.UNAVAILABLE, null);
-  }
-
-  @Override
-  public String toString() {
-    return "IdempotencyStartOutcome[status=" + status + ", operationId=REDACTED]";
-  }
-
-  private static String validateOperation(Status status, String operationId) {
-    if (status == Status.INDETERMINATE) {
-      return IdempotencyV2Validation.opaqueToken(operationId, "operationId");
-    }
-    if (operationId != null) {
-      throw new IllegalArgumentException("operationId is valid only for INDETERMINATE");
-    }
-    return null;
-  }
-
-  public enum Status {
-    STARTED,
-    ALREADY_STARTED_SAME_OPERATION,
-    ABSENT,
-    NOT_OWNER,
-    NOT_CLAIMED,
-    OPERATION_CONFLICT,
-    INDETERMINATE,
-    UNAVAILABLE
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyStorePortV2.java

@@ -1,35 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.time.Duration;
-
-/**
- * Owner-safe request-replay store contract.
- *
- * <p>This contract does not promise cross-store exactly-once. Every mutation compares the owner
- * token and attempt, and every uncertain response remains inspectable with the caller-retained
- * operation token.
- */
-public interface IdempotencyStorePortV2 {
-
-  IdempotencyClaimAttempt newClaimAttempt(String operationId);
-
-  IdempotencyClaimOutcome claim(IdempotencyClaimRequest request);
-
-  IdempotencyStartOutcome markExecutionStarted(IdempotencyOwner owner, String operationId);
-
-  IdempotencyRenewOutcome renew(
-      IdempotencyOwner owner, Duration processingLeaseTtl, String operationId);
-
-  IdempotencyCompleteOutcome complete(
-      IdempotencyOwner owner, StoredResponse response, Duration replayTtl, String operationId);
-
-  IdempotencyFailOutcome markFailed(
-      IdempotencyOwner owner,
-      IdempotencyFailureDisposition disposition,
-      Duration retention,
-      String operationId);
-
-  IdempotencyReleaseOutcome releaseBeforeExecution(IdempotencyOwner owner, String operationId);
-
-  IdempotencyInspection inspect(IdempotencyInspectionRequest request);
-}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/IdempotencyV2Validation.java

@@ -1,50 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.util.Objects;
-
-final class IdempotencyV2Validation {
-
-  static final Duration MAXIMUM_PROCESSING_LEASE = Duration.ofHours(24);
-  static final Duration MAXIMUM_REPLAY_TTL = Duration.ofDays(30);
-  static final Duration MAXIMUM_RETRY_AFTER = Duration.ofMinutes(5);
-
-  private IdempotencyV2Validation() {}
-
-  static String opaqueToken(String value, String field) {
-    if (value == null
-        || value.length() < 16
-        || value.length() > 128
-        || !value.matches("[A-Za-z0-9_-]+")) {
-      throw new IllegalArgumentException(field + " must contain 16..128 Base64URL-safe characters");
-    }
-    return value;
-  }
-
-  static String boundedId(String value, String field) {
-    if (value == null || !value.matches("[a-z][a-z0-9._-]{0,62}")) {
-      throw new IllegalArgumentException(field + " must be a bounded identifier");
-    }
-    return value;
-  }
-
-  static Duration positiveBounded(Duration value, Duration maximum, String field) {
-    Objects.requireNonNull(value, field + " must be non-null");
-    if (value.isZero() || value.isNegative() || value.compareTo(maximum) > 0) {
-      throw new IllegalArgumentException(field + " must be positive and bounded");
-    }
-    return value;
-  }
-
-  static Instant instant(Instant value, String field) {
-    return Objects.requireNonNull(value, field + " must be non-null");
-  }
-
-  static long positiveAttempt(long value, String field) {
-    if (value < 1 || value > 1_000_000_000L) {
-      throw new IllegalArgumentException(field + " must be in 1..1000000000");
-    }
-    return value;
-  }
-}

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/callback/DeliveryAttemptResolverPort.java

@@ -17,4 +17,18 @@ public interface DeliveryAttemptResolverPort {
    */
   Optional<DeliveryAttemptSnapshot> byProviderRequestId(
       ProviderProfileId profileId, String providerRequestId);
+
+  /**
+   * Look up by the hash a stored event already carries.
+   *
+   * <p>The overload above needs the identifier in clear, which only an incoming callback has. A
+   * stored event has the hash and nothing else, so without this lookup a sweep over unmatched
+   * events had no matcher at all once the attempt id was absent — and an event that arrived before
+   * its attempt was written stayed unmatched for good.
+   *
+   * <p>Scoped to the profile as well as the hash: two profiles can mint the same provider request
+   * id, and matching on the hash alone would attach one tenant's callback to another's attempt.
+   */
+  Optional<DeliveryAttemptSnapshot> byProviderRequestIdHash(
+      ProviderProfileId profileId, ProviderRequestIdHash providerRequestIdHash);
 }

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/callback/ProviderEventLedger.java

@@ -55,6 +55,24 @@ public interface ProviderEventLedger {
       String providerRequestId,
       dev.caskeleton.application.notification.platform.api.DeliveryAttemptId attemptId);
 
+  /**
+   * Binds one already-stored event to the attempt a sweep has just matched it to.
+   *
+   * <p>{@link #bindUnmatched} is the dispatch side of the same race: the sending process finishes,
+   * learns the provider request id and goes looking for callbacks that beat it. This is the sweep
+   * side, for the events that call never reached — the dispatching process died between writing the
+   * attempt and binding, or the bind itself failed. Without it, matching an event during a sweep
+   * left the row's {@code attempt_id} null and the next pass had to match it all over again.
+   *
+   * <p>Conditional on the event still being unbound, so a sweep cannot move an event another
+   * matcher has already claimed.
+   *
+   * @return whether this call is the one that bound it
+   */
+  boolean bindAttempt(
+      ProviderEventRecordId eventId,
+      dev.caskeleton.application.notification.platform.api.DeliveryAttemptId attemptId);
+
   /** Every stored event for one attempt, oldest first, for projection replay. */
   List<ProviderEventRecord> eventsForAttempt(
       dev.caskeleton.application.notification.platform.api.DeliveryAttemptId attemptId);

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/callback/ProviderEventProjectionService.java

@@ -85,16 +85,39 @@ public final class ProviderEventProjectionService {
             }));
   }
 
+  /**
+   * Finds the attempt an event belongs to, and records the answer.
+   *
+   * <p>The last matcher is the one that makes a sweep work at all. The raw provider request id is
+   * present only on an incoming callback and is never stored, so a rehydrated event offers nothing
+   * but its hash — and an event that arrived before its attempt existed has no attempt id either.
+   * Between them, a replayed event had no matcher and stayed {@code PENDING} on every pass.
+   *
+   * <p>A match found by hash is bound to the row, so the next pass resolves it by identity instead
+   * of hashing its way back to the same attempt.
+   */
   private Optional<DeliveryAttemptSnapshot> resolve(ProviderEventRecord event) {
     Optional<DeliveryAttemptSnapshot> byAttempt =
         event.attemptId().flatMap(attemptResolver::byAttemptId);
     if (byAttempt.isPresent()) {
       return byAttempt;
     }
-    return event
-        .event()
-        .providerRequestId()
-        .flatMap(
-            requestId -> attemptResolver.byProviderRequestId(event.providerProfileId(), requestId));
+    Optional<DeliveryAttemptSnapshot> byRequestId =
+        event
+            .event()
+            .providerRequestId()
+            .flatMap(
+                requestId ->
+                    attemptResolver.byProviderRequestId(event.providerProfileId(), requestId));
+    Optional<DeliveryAttemptSnapshot> matched =
+        byRequestId.isPresent()
+            ? byRequestId
+            : event
+                .providerRequestIdHash()
+                .flatMap(
+                    hash ->
+                        attemptResolver.byProviderRequestIdHash(event.providerProfileId(), hash));
+    matched.ifPresent(attempt -> ledger.bindAttempt(event.id(), attempt.attemptId()));
+    return matched;
   }
 }

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/callback/ProviderEventRecord.java

@@ -6,12 +6,21 @@ import java.time.Instant;
 import java.util.Objects;
 import java.util.Optional;
 
-/** A stored, immutable ledger entry. */
+/**
+ * A stored, immutable ledger entry.
+ *
+ * <p>{@code providerRequestIdHash} is how an unbound entry finds its attempt later. The raw
+ * provider request id lives only on the incoming callback and is never stored, so a rehydrated
+ * entry's {@code event().providerRequestId()} is always empty — which left the sweep with nothing
+ * but the attempt id to match on, and an event that arrived before its attempt has no attempt id.
+ * Such an event stayed {@code PENDING} for good.
+ */
 public record ProviderEventRecord(
     ProviderEventRecordId id,
     ProviderProfileId providerProfileId,
     NormalizedProviderEvent event,
     Optional<DeliveryAttemptId> attemptId,
+    Optional<ProviderRequestIdHash> providerRequestIdHash,
     ProviderEventSource source,
     boolean signatureVerified,
     Instant receivedAt,
@@ -25,6 +34,7 @@ public record ProviderEventRecord(
     Objects.requireNonNull(providerProfileId, "providerProfileId");
     Objects.requireNonNull(event, "event");
     Objects.requireNonNull(attemptId, "attemptId");
+    Objects.requireNonNull(providerRequestIdHash, "providerRequestIdHash");
     Objects.requireNonNull(source, "source");
     Objects.requireNonNull(receivedAt, "receivedAt");
     Objects.requireNonNull(rawPayloadDigest, "rawPayloadDigest");

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/callback/ProviderRequestIdHash.java

@@ -0,0 +1,38 @@
+package dev.caskeleton.application.notification.platform.callback;
+
+import java.util.Locale;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * The keyed hash of a provider's own request id, as the ledger stored it.
+ *
+ * <p>A stored event used to carry no way back to its attempt. The raw provider request id is
+ * deliberately never persisted — it would make the attempt table a readable index of every message
+ * the platform sent — and rehydrating a row therefore produced a record whose {@code
+ * providerRequestId} was always empty. So the only matcher that could run after ingestion was the
+ * attempt id, and an event that arrived before its attempt existed had none: the sweep looked at it
+ * on every pass, found nothing to match on, and left it {@code PENDING} forever.
+ *
+ * <p>Carrying the hash is what makes the late match possible without storing the identifier. The
+ * value is bounded because it reaches queries and diagnostics: an unbounded string from a stored
+ * row is input, not a fact.
+ */
+public record ProviderRequestIdHash(String value) {
+
+  /** Hex, and long enough that a truncated digest cannot be mistaken for one. */
+  private static final Pattern FORMAT = Pattern.compile("[0-9a-f]{32,128}");
+
+  public ProviderRequestIdHash {
+    Objects.requireNonNull(value, "value");
+    value = value.toLowerCase(Locale.ROOT);
+    if (!FORMAT.matcher(value).matches()) {
+      throw new IllegalArgumentException("provider request id hash must be a bounded hex digest");
+    }
+  }
+
+  @Override
+  public String toString() {
+    return value;
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/dispatch/NotificationDispatchService.java

@@ -173,7 +173,7 @@ public final class NotificationDispatchService {
             () -> recorder.record(attempt, work.recipient(), result, clock.instant()));
 
     RetryDecision next = retryPolicy.decide(retryContext(work, recorded, result, profile));
-    applyNextAction(work, recorded, next, clock.instant());
+    applyNextAction(work, recorded, next, clock.instant(), lease);
     releaseLease(lease);
   }
 
@@ -398,30 +398,48 @@ public final class NotificationDispatchService {
         });
   }
 
+  /**
+   * Applies the outcome, and only while this worker still holds the job.
+   *
+   * <p>Every write here happens *after* the provider call, which is the one stretch the platform
+   * deliberately spends outside a transaction. A lease can expire during it, another worker can
+   * claim the job, and this worker can then wake up and describe an attempt that is no longer the
+   * live one. The fenced variants make that write match nothing rather than win.
+   *
+   * <p>Losing the lease is not an error to report. The new holder owns the job and will record its
+   * own outcome; this worker's only remaining obligation is to stop.
+   */
   private void applyNextAction(
-      Work work, DeliveryAttemptRecord attempt, RetryDecision decision, Instant now) {
+      Work work,
+      DeliveryAttemptRecord attempt,
+      RetryDecision decision,
+      Instant now,
+      RecipientLease lease) {
     transactions.inWrite(
         () -> {
           switch (decision) {
             case RetryDecision.RetryAfter retry ->
-                recipients.transition(
+                recipients.transitionHeldBy(
                     work.recipient().id(),
                     RecipientDeliveryState.RETRY_WAITING,
-                    Optional.of(now.plus(retry.delay())));
+                    Optional.of(now.plus(retry.delay())),
+                    lease);
             case RetryDecision.Reconcile reconcile ->
-                recipients.transition(
+                recipients.transitionHeldBy(
                     work.recipient().id(),
                     RecipientDeliveryState.RECONCILIATION_REQUIRED,
-                    Optional.of(reconcile.at()));
+                    Optional.of(reconcile.at()),
+                    lease);
             case RetryDecision.Fallback ignored ->
-                recipients.save(advanceRoute(work.recipient(), now));
+                recipients.saveHeldBy(advanceRoute(work.recipient(), now), lease);
             case RetryDecision.Stop ignored ->
-                recipients.transition(
+                recipients.transitionHeldBy(
                     work.recipient().id(),
                     attempt.submissionOutcome() == SubmissionOutcome.CONFIRMED_ACCEPTED
                         ? RecipientDeliveryState.COMPLETED
                         : RecipientDeliveryState.FAILED,
-                    Optional.empty());
+                    Optional.empty(),
+                    lease);
           }
           refreshStatus(work);
         });

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/dispatch/RecipientDeliveryStorePort.java

@@ -17,4 +17,36 @@ public interface RecipientDeliveryStorePort {
   /** Move a job to a state with an optional next dispatch time. */
   RecipientDeliveryRecord transition(
       RecipientDeliveryId id, RecipientDeliveryState state, Optional<Instant> nextDispatchAt);
+
+  /**
+   * Stores a modified job only while the given lease still holds it.
+   *
+   * <p>The claim is fenced and the renew is fenced, and for a while the *completion* was not. A
+   * worker whose lease expired during a provider call — the one place the platform deliberately
+   * spends time outside a transaction — came back and wrote its outcome with an unconditional
+   * {@code save}, over the row a new holder had already claimed and might already have dispatched.
+   * The optimistic {@code version} column did not help: it detects a concurrent edit, not a
+   * superseded writer, and the late worker's read was recent enough to win.
+   *
+   * @param record the modified job
+   * @param lease the lease the caller believes it holds
+   * @return the stored job, or empty when the lease has been superseded and nothing was written
+   */
+  Optional<RecipientDeliveryRecord> saveHeldBy(
+      RecipientDeliveryRecord record, RecipientLease lease);
+
+  /**
+   * Moves a job to a state only while the given lease still holds it.
+   *
+   * @param id the job
+   * @param state the state to move to
+   * @param nextDispatchAt when it next becomes due, when it does
+   * @param lease the lease the caller believes it holds
+   * @return the stored job, or empty when the lease has been superseded and nothing was written
+   */
+  Optional<RecipientDeliveryRecord> transitionHeldBy(
+      RecipientDeliveryId id,
+      RecipientDeliveryState state,
+      Optional<Instant> nextDispatchAt,
+      RecipientLease lease);
 }

src/application-core/src/main/java/dev/caskeleton/application/notification/platform/security/SecretPurpose.java

@@ -1,13 +1,18 @@
 package dev.caskeleton.application.notification.platform.security;
 
-/** Key purposes. A key issued for one purpose is never reused for another. */
+/**
+ * Key purposes. A key issued for one purpose is never reused for another.
+ *
+ * <p>Every constant here has a consumer. An {@code UNSUBSCRIBE_TOKEN} purpose sat in this enum with
+ * no binding in configuration and no code that asked for it, so startup validation demanded nothing
+ * for it and nothing could have used it if it had: an entry that only looked like coverage.
+ */
 public enum SecretPurpose {
   CONTACT_ENCRYPTION,
   CONTACT_LOOKUP_HMAC,
   CALLBACK_SIGNING,
   PROVIDER_CREDENTIAL,
   VAPID_SIGNING,
-  UNSUBSCRIBE_TOKEN,
   PAYLOAD_ENCRYPTION,
 
   /**

src/application-core/src/main/java/dev/caskeleton/application/outbox/NewOutboxEvent.java

@@ -32,6 +32,14 @@ public record NewOutboxEvent(
     requireNonBlank(eventType, "eventType");
     requireNonBlank(aggregateId, "aggregateId");
     requireNonBlank(payload, "payload");
+    // Validated here, on the append side, and not only on the read model.
+    //
+    // OutboxEvent — the type the relay maps a stored row into — already applied this policy, so a
+    // malformed or oversized payload was accepted at write and first threw when the relay read it
+    // back. And it threw out of `claimBatch`, which maps a whole batch in one stream: one poisoned
+    // row failed every subsequent relay pass rather than the single transaction that could still
+    // have refused it. The append boundary is where the caller is still there to be told.
+    OutboxPayloadPolicy.requireValidPayload(payload);
     Objects.requireNonNull(occurredAt, "occurredAt must not be null");
     requireNonBlank(correlationId, "correlationId");
     requireNonBlank(idempotencyKey, "idempotencyKey");

src/application-core/src/main/java/dev/caskeleton/application/outbox/OutboxPublishAmbiguousException.java

@@ -0,0 +1,24 @@
+package dev.caskeleton.application.outbox;
+
+/**
+ * What the relay reports when a publish neither succeeded nor definitely failed.
+ *
+ * <p>The message left this process and no acknowledgement arrived, so it may be stored in the
+ * broker. The row stays claimable, which means a retry may duplicate — and that is the deliberate
+ * choice: the alternative loses a message that was in fact delivered. Naming the state in the
+ * report is what lets an operator see duplicates coming instead of finding them.
+ *
+ * <p>Never thrown across the port. It exists only to carry a returned outcome into a report.
+ */
+public final class OutboxPublishAmbiguousException extends RuntimeException {
+
+  private static final long serialVersionUID = 1L;
+
+  /** Creates the cause for an ambiguous publish. */
+  public OutboxPublishAmbiguousException() {
+    super(
+        "the broker accepted this event's frame and never confirmed it; the row stays claimable and"
+            + " a retry may duplicate, which is the bounded risk rather than losing a delivered"
+            + " message");
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/OutboxPublishRefusedException.java

@@ -0,0 +1,43 @@
+package dev.caskeleton.application.outbox;
+
+/**
+ * What the relay reports when a publish was refused without throwing.
+ *
+ * <p>An adapter that returns {@link OutboxPublishOutcome#REJECTED_BEFORE_SEND} or {@link
+ * OutboxPublishOutcome#REJECTED_AFTER_BROKER} has already decided the attempt is over and has no
+ * exception to hand up. The failure report needs a cause, and inventing a generic one would put a
+ * stack trace from this class in an operator's hands with nothing in it about the refusal — so the
+ * refusal itself is the cause, and it says which of the two it was.
+ *
+ * <p>Never thrown across the port. It exists only to carry a returned outcome into a report.
+ */
+public final class OutboxPublishRefusedException extends RuntimeException {
+
+  private static final long serialVersionUID = 1L;
+
+  private final transient OutboxPublishOutcome outcome;
+
+  /**
+   * Creates the cause for a refused publish.
+   *
+   * @param outcome the refusal the adapter reported
+   */
+  public OutboxPublishRefusedException(OutboxPublishOutcome outcome) {
+    super(
+        outcome == OutboxPublishOutcome.REJECTED_BEFORE_SEND
+            ? "the broker adapter refused this event before transmitting it, so nothing was stored"
+                + " anywhere and the same payload will be refused again"
+            : "the broker received this event and refused it; the cause is broker-side policy"
+                + " rather than this application's routing");
+    this.outcome = outcome;
+  }
+
+  /**
+   * The refusal this cause carries.
+   *
+   * @return the reported outcome
+   */
+  public OutboxPublishOutcome outcome() {
+    return outcome;
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/PublishPendingOutboxEventsUseCase.java

@@ -115,12 +115,27 @@ public class PublishPendingOutboxEventsUseCase
    * timeout. See README for both failure modes.
    */
   private OutboxRelayResult.Outcome publishOne(OutboxEvent event, Instant now) {
+    OutboxPublishOutcome achieved;
     try {
-      publishPort.publish(event);
+      // The four-valued call, not the throwing one. An adapter that cannot tell an ambiguous
+      // publish apart still throws and is handled below exactly as before; one that can no longer
+      // has to pretend a message it may have stored was definitely not stored.
+      achieved = publishPort.publishForOutcome(event);
     } catch (RuntimeException publishEx) {
       // Publish failure: drive FAILED/DEAD state machine + typed report; do NOT rethrow.
       return handlePublishFailure(event, now, publishEx);
     }
+    if (achieved == OutboxPublishOutcome.AMBIGUOUS) {
+      // The message left this process and no acknowledgement arrived, so it may be stored. Treated
+      // as retryable and reported as such: the duplicate risk is the caller's to bound with an
+      // idempotency key, and the alternative — parking it — loses a message that was delivered.
+      return handlePublishFailure(event, now, new OutboxPublishAmbiguousException());
+    }
+    if (achieved != OutboxPublishOutcome.CONFIRMED) {
+      // Definite refusal. Retrying the same payload fails the same way, so the attempt budget is
+      // not spent on it; the row is dead-lettered now rather than after the timer runs out.
+      return deadLetter(event, new OutboxPublishRefusedException(achieved));
+    }
     // markPublished failure (if any) propagates: the row stays IN_FLIGHT and is
     // recovered via the orphan visibility-timeout reclaim path.
     tx.inWrite(() -> store.markPublished(event.eventId()));
@@ -137,17 +152,7 @@ public class PublishPendingOutboxEventsUseCase
 
     if (event.attemptCount() >= backoffPolicy.maxAttempts()) {
       // All attempts exhausted — DEAD-letter the event.
-      tx.inWrite(() -> store.markDead(event.eventId()));
-      reportFailure(
-          () ->
-              OutboxRelayFailureReport.deadLetter(
-                  event.eventId(),
-                  event.eventType(),
-                  event.aggregateId(),
-                  event.correlationId(),
-                  event.attemptCount(),
-                  cause));
-      return OutboxRelayResult.Outcome.DEAD;
+      return deadLetter(event, cause);
     } else {
       // Transient failure — schedule retry with exponential backoff.
       Instant nextAttemptAt = backoffPolicy.nextAttemptAt(event.attemptCount(), now);
@@ -166,6 +171,20 @@ public class PublishPendingOutboxEventsUseCase
     }
   }
 
+  private OutboxRelayResult.Outcome deadLetter(OutboxEvent event, RuntimeException cause) {
+    tx.inWrite(() -> store.markDead(event.eventId()));
+    reportFailure(
+        () ->
+            OutboxRelayFailureReport.deadLetter(
+                event.eventId(),
+                event.eventType(),
+                event.aggregateId(),
+                event.correlationId(),
+                event.attemptCount(),
+                cause));
+    return OutboxRelayResult.Outcome.DEAD;
+  }
+
   private void reportFailure(Supplier<OutboxRelayFailureReport> reportFactory) {
     try {
       failureReporter.report(reportFactory.get());

src/application-core/src/main/java/dev/caskeleton/application/security/ObjectAccessDecision.java

@@ -0,0 +1,43 @@
+package dev.caskeleton.application.security;
+
+/**
+ * The answer to one object-access question.
+ *
+ * <p>A denial carries a stable code rather than a message, so a caller can branch on it and a
+ * metric can count it without the reason itself disclosing what exists.
+ *
+ * <p>{@code hideExistence} is a separate flag because "you may not see this" and "there is nothing
+ * here" are different answers to the client and the same answer to the application: only the
+ * application knows whether admitting the object exists is itself a disclosure, so the transport is
+ * told rather than left to guess.
+ *
+ * @param allowed whether the caller may proceed
+ * @param code stable denial code, or {@code null} when allowed
+ * @param hideExistence whether a denial must read as "not found" rather than "forbidden"
+ */
+public record ObjectAccessDecision(boolean allowed, String code, boolean hideExistence) {
+
+  public ObjectAccessDecision {
+    if (!allowed && (code == null || code.isBlank())) {
+      throw new IllegalArgumentException("a denial must carry a stable code");
+    }
+    if (allowed && code != null) {
+      throw new IllegalArgumentException("an allowed decision carries no denial code");
+    }
+  }
+
+  /** The caller may proceed. */
+  public static ObjectAccessDecision allow() {
+    return new ObjectAccessDecision(true, null, false);
+  }
+
+  /** The caller is denied, and the denial is visible as such. */
+  public static ObjectAccessDecision deny(String code) {
+    return new ObjectAccessDecision(false, code, false);
+  }
+
+  /** The caller is denied and the object's existence is hidden. */
+  public static ObjectAccessDecision denyHidingExistence(String code) {
+    return new ObjectAccessDecision(false, code, true);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/security/ObjectAccessPolicy.java

@@ -0,0 +1,54 @@
+package dev.caskeleton.application.security;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Decides whether a caller may see a particular object.
+ *
+ * <p>Distinct from {@link AuthorizationPort}, which answers "may this caller perform this kind of
+ * operation at all?" from the principal's permissions alone. This one needs the object: two callers
+ * with the same permission get different answers about the same row, and the difference is domain
+ * state rather than a role.
+ *
+ * <p>Owned here rather than by the transport that asks. An inbound adapter that declared this
+ * contract would force every implementation to compile against that adapter — the dependency
+ * inversion this layer exists to prevent — and would make the same rule unavailable to the next
+ * caller that needs it.
+ */
+public interface ObjectAccessPolicy {
+
+  /**
+   * Decides access to one object.
+   *
+   * @param request the object and the caller asking about it
+   */
+  ObjectAccessDecision decide(ObjectAccessRequest request);
+
+  /**
+   * Decides access to many objects of one type at once.
+   *
+   * <p>The batch form exists because per-object authorization inside a batched load reintroduces
+   * exactly the N+1 the batching removed. The default answers one at a time; an implementation with
+   * a set-based rule overrides it with a single query.
+   *
+   * @param actorId the acting identity, or {@code null} for an unauthenticated caller
+   * @param tenantId the tenant the objects are asked about in
+   * @param objectType the type name of the objects
+   * @param objectIds external identities, in the order they were requested
+   * @return a decision for every requested id, in the requested order
+   */
+  default Map<String, ObjectAccessDecision> decideAll(
+      String actorId, String tenantId, String objectType, List<String> objectIds) {
+
+    Map<String, ObjectAccessDecision> decisions = new LinkedHashMap<>();
+    objectIds.forEach(
+        objectId ->
+            decisions.put(
+                objectId,
+                decide(new ObjectAccessRequest(actorId, tenantId, objectType, objectId))));
+    return Collections.unmodifiableMap(decisions);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/security/ObjectAccessRequest.java

@@ -0,0 +1,39 @@
+package dev.caskeleton.application.security;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * A question about one object, asked in terms no transport owns.
+ *
+ * <p>Object-level access depends on domain state — ownership, membership, workflow status — so the
+ * answer belongs to the application. What reaches it must not, because a contract phrased in a
+ * transport's own request type can only be implemented by code that compiles against that
+ * transport: a GraphQL request context here would make this package depend on the inbound GraphQL
+ * adapter, invert the dependency for every persistence adapter the use case reaches, and leave a
+ * REST or scheduled caller unable to ask the same question at all.
+ *
+ * <p>Four strings, therefore. Each caller maps its own context down to them.
+ *
+ * @param actorId the acting identity, or {@code null} for an unauthenticated caller
+ * @param tenantId the tenant the object is asked about in
+ * @param objectType the type name of the object
+ * @param objectId external identity of the object
+ */
+public record ObjectAccessRequest(
+    String actorId, String tenantId, String objectType, String objectId) {
+
+  public ObjectAccessRequest {
+    Objects.requireNonNull(tenantId, "tenant is required");
+    Objects.requireNonNull(objectType, "object type is required");
+    Objects.requireNonNull(objectId, "object id is required");
+    if (tenantId.isBlank() || objectType.isBlank() || objectId.isBlank()) {
+      throw new IllegalArgumentException("tenant, object type and object id are required");
+    }
+  }
+
+  /** The acting identity, absent for an unauthenticated caller. */
+  public Optional<String> actor() {
+    return Optional.ofNullable(actorId);
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/fileserver/api/metadata/MetadataPortContractTest.java

@@ -149,6 +149,7 @@ class MetadataPortContractTest {
             Optional.empty(),
             Optional.empty(),
             Optional.empty(),
+            false,
             0,
             NOW,
             NOW);

src/application-core/src/test/java/dev/caskeleton/application/fileserver/cleanup/CleanupServiceTest.java

@@ -1,11 +1,13 @@
 package dev.caskeleton.application.fileserver.cleanup;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import dev.caskeleton.application.fileserver.api.ContentKey;
 import dev.caskeleton.application.fileserver.api.FileId;
 import dev.caskeleton.application.fileserver.api.FileState;
 import dev.caskeleton.application.fileserver.api.UploadId;
+import dev.caskeleton.application.fileserver.api.error.ConcurrentFileModificationException;
 import dev.caskeleton.application.fileserver.api.metadata.FileRecord;
 import dev.caskeleton.application.fileserver.api.metadata.UploadSession;
 import dev.caskeleton.application.fileserver.api.metadata.UploadSessionDraft;
@@ -115,6 +117,8 @@ class CleanupServiceTest {
             UploadProtocol.RAW,
             OptionalLong.of(SIZE),
             FileserverFixtures.NOW.plusSeconds(3600)));
+    // What cancel does before it queues the item, in the same transaction.
+    sessions.terminate(uploadId);
     content.storeStaging(uploadId);
     queue.enqueue(CleanupRequest.forStaging(CleanupType.CANCELLED_STAGING, fileId, uploadId));
 
@@ -124,6 +128,118 @@ class CleanupServiceTest {
     assertThat(content.stagingExists(uploadId)).isFalse();
   }
 
+  /**
+   * The upload has to be finished with, not merely unleased.
+   *
+   * <p>Cleanup used to delete on the strength of the lease alone, and an upload between two appends
+   * holds no lease at all. Its staged bytes were then indistinguishable from a cancelled upload's,
+   * so a queue item naming a live upload removed the object it was writing.
+   */
+  @Test
+  void cleanupDoesNotDeleteTheStagingObjectOfAnUploadStillOpenForWriting() {
+    UploadId uploadId = UploadId.of(UUID.randomUUID());
+    FileId fileId = FileId.of(UUID.randomUUID());
+    sessions.create(
+        new UploadSessionDraft(
+            uploadId,
+            fileId,
+            UploadProtocol.RAW,
+            OptionalLong.of(SIZE),
+            FileserverFixtures.NOW.plusSeconds(3600)));
+    content.storeStaging(uploadId);
+    queue.enqueue(CleanupRequest.forStaging(CleanupType.CANCELLED_STAGING, fileId, uploadId));
+
+    CleanupBatchResult result = cleanup.runBatch(100, 1L << 30);
+
+    assertThat(result.skippedActiveLease()).isEqualTo(1);
+    assertThat(content.stagingExists(uploadId)).isTrue();
+  }
+
+  /**
+   * The barrier: exactly one of the writer and the cleanup wins.
+   *
+   * <p>A writer that already holds the lease when the item is processed keeps the bytes; once that
+   * lease lapses the cleanup claims the session, and the writer's own acquire is then refused
+   * because the session is terminal. There is no ordering in which both succeed.
+   */
+  @Test
+  void aWriterCannotTakeALeaseOnAnUploadCleanupHasClaimed() {
+    UploadId uploadId = UploadId.of(UUID.randomUUID());
+    FileId fileId = FileId.of(UUID.randomUUID());
+    UploadSession created =
+        sessions.create(
+            new UploadSessionDraft(
+                uploadId,
+                fileId,
+                UploadProtocol.RAW,
+                OptionalLong.of(SIZE),
+                FileserverFixtures.NOW.plusSeconds(3600)));
+    sessions.terminate(uploadId);
+    content.storeStaging(uploadId);
+    queue.enqueue(CleanupRequest.forStaging(CleanupType.CANCELLED_STAGING, fileId, uploadId));
+
+    CleanupBatchResult result = cleanup.runBatch(100, 1L << 30);
+
+    assertThat(result.deleted()).isEqualTo(1);
+    assertThatThrownBy(
+            () ->
+                sessions.acquireLease(
+                    uploadId,
+                    "node-b",
+                    FileserverFixtures.NOW,
+                    Duration.ofSeconds(30),
+                    created.version()))
+        .as("granting this lease would hand a writer an object that no longer exists")
+        .isInstanceOf(ConcurrentFileModificationException.class);
+  }
+
+  /**
+   * Abandoned claims come back before the batch looks for work.
+   *
+   * <p>A claim takes an item out of the due set, so a worker that performed the physical delete and
+   * then died left its item IN_PROGRESS where no later batch could see it. The reclaim query
+   * existed and nothing called it, which made the recovery path a comment. Sweeping first also
+   * means a recovered item is eligible in the same pass rather than the next one.
+   */
+  @Test
+  void everyBatchReclaimsAbandonedClaimsBeforeClaimingNewWork() {
+    cleanup.runBatch(100, 1L << 30);
+
+    assertThat(queue.reclaimSweeps()).hasSize(1);
+    assertThat(queue.claimSweeps()).hasSize(1);
+    assertThat(queue.reclaimSweeps().get(0))
+        .as("reclaiming after the claim leaves recovered items for the next batch")
+        .isEqualTo(queue.claimSweeps().get(0));
+  }
+
+  /**
+   * Settling twice reclaims once.
+   *
+   * <p>A worker that deleted the staged bytes and died before the queue item was settled leaves the
+   * item to be retried. The claim is written so the retry matches an already-cleared lease rather
+   * than finding nothing to claim and giving up on an item whose object is already gone.
+   */
+  @Test
+  void aRetryAfterTheDeleteSucceededSettlesTheItemAgain() {
+    UploadId uploadId = UploadId.of(UUID.randomUUID());
+    FileId fileId = FileId.of(UUID.randomUUID());
+    sessions.create(
+        new UploadSessionDraft(
+            uploadId,
+            fileId,
+            UploadProtocol.RAW,
+            OptionalLong.of(SIZE),
+            FileserverFixtures.NOW.plusSeconds(3600)));
+    sessions.terminate(uploadId);
+    content.storeStaging(uploadId);
+    queue.enqueue(CleanupRequest.forStaging(CleanupType.CANCELLED_STAGING, fileId, uploadId));
+    cleanup.runBatch(100, 1L << 30);
+
+    assertThat(sessions.claimForCleanup(uploadId, FileserverFixtures.NOW))
+        .as("a claim that answers no on the retry strands an item whose object is already deleted")
+        .isTrue();
+  }
+
   @Test
   void anItemWhoseRecordWasRepublishedUnderAnotherKeyIsDiscardedNotExecuted() {
     FileRecord deleting = deletingRecord();

src/application-core/src/test/java/dev/caskeleton/application/fileserver/testkit/FileserverFixtures.java

@@ -43,6 +43,7 @@ public final class FileserverFixtures {
         Optional.of("node-a"),
         Optional.of(UUID.randomUUID()),
         Optional.of(NOW.plusSeconds(30)),
+        false,
         0,
         NOW,
         NOW);

src/application-core/src/test/java/dev/caskeleton/application/fileserver/testkit/InMemoryCleanupQueue.java

@@ -13,21 +13,37 @@ import java.util.UUID;
 /**
  * Hand-rolled cleanup queue that records what was scheduled.
  *
- * <p>Like the durable queue, it assigns the item identity on enqueue.
+ * <p>Like the durable queue, it assigns the item identity on enqueue. Claim leases are recorded
+ * rather than enforced: the fencing they exist for is a database property, and a fake that pretends
+ * to have it would report success for a guarantee only real PostgreSQL can give.
  */
 public final class InMemoryCleanupQueue implements CleanupQueue {
 
   private final List<CleanupItem> queued = new ArrayList<>();
   private final List<CleanupItem> done = new ArrayList<>();
   private final Map<CleanupItem, String> failed = new LinkedHashMap<>();
+  private final List<Instant> reclaimSweeps = new ArrayList<>();
+  private final List<Instant> claimSweeps = new ArrayList<>();
 
   @Override
   public void enqueue(CleanupRequest request) {
     queued.add(new CleanupItem(UUID.randomUUID(), request, 0));
   }
 
+  @Override
+  public int reclaimExpiredClaims(Instant now, int limit) {
+    reclaimSweeps.add(now);
+    return 0;
+  }
+
+  /** The moments a batch asked for abandoned claims back, in order. */
+  public List<Instant> reclaimSweeps() {
+    return List.copyOf(reclaimSweeps);
+  }
+
   @Override
   public List<CleanupItem> claimDue(Instant now, int limit) {
+    claimSweeps.add(now);
     return List.copyOf(queued.subList(0, Math.min(limit, queued.size())));
   }
 
@@ -42,6 +58,11 @@ public final class InMemoryCleanupQueue implements CleanupQueue {
     failed.put(item, reasonCode);
   }
 
+  /** The moments a batch claimed due items, in order. */
+  public List<Instant> claimSweeps() {
+    return List.copyOf(claimSweeps);
+  }
+
   public List<CleanupItem> queued() {
     return List.copyOf(queued);
   }

src/application-core/src/test/java/dev/caskeleton/application/fileserver/testkit/InMemoryUploadSessionStore.java

@@ -21,7 +21,8 @@ import java.util.UUID;
  * Hand-rolled session store with the same lease semantics as the JPA adapter.
  *
  * <p>A lease is granted only when none is held or the held one expired, and an offset commit
- * requires the exact token plus the expected offset.
+ * requires the exact token plus the expected offset. A terminal session refuses all three, which is
+ * what the {@code lifecycle_state = 'ACTIVE'} clause on the JPA statements does.
  */
 public final class InMemoryUploadSessionStore implements UploadSessionStore {
 
@@ -45,6 +46,7 @@ public final class InMemoryUploadSessionStore implements UploadSessionStore {
             Optional.empty(),
             Optional.empty(),
             Optional.empty(),
+            false,
             0,
             now,
             now);
@@ -67,7 +69,7 @@ public final class InMemoryUploadSessionStore implements UploadSessionStore {
     UploadSession current = require(uploadId);
     boolean leaseFree =
         current.leaseUntil().isEmpty() || !current.leaseUntil().get().isAfter(clockNow);
-    if (current.version() != expectedVersion || !leaseFree) {
+    if (current.terminal() || current.version() != expectedVersion || !leaseFree) {
       throw new ConcurrentFileModificationException(
           "writer lease is held by another owner",
           FileserverFailureContext.forUpload(
@@ -92,7 +94,7 @@ public final class InMemoryUploadSessionStore implements UploadSessionStore {
     UploadSession current = require(lease.uploadId());
     boolean tokenMatches = current.leaseToken().map(lease.token()::equals).orElse(false);
     boolean stillHeld = current.leaseUntil().map(clockNow::isBefore).orElse(false);
-    if (!tokenMatches || !stillHeld) {
+    if (current.terminal() || !tokenMatches || !stillHeld) {
       throw new ConcurrentFileModificationException(
           "writer lease can no longer be renewed",
           FileserverFailureContext.forUpload(
@@ -116,7 +118,7 @@ public final class InMemoryUploadSessionStore implements UploadSessionStore {
       UploadId uploadId, WriterLease lease, long expectedOffset, long committedOffset) {
     UploadSession current = require(uploadId);
     boolean tokenMatches = current.leaseToken().map(lease.token()::equals).orElse(false);
-    if (!tokenMatches || current.committedOffset() != expectedOffset) {
+    if (current.terminal() || !tokenMatches || current.committedOffset() != expectedOffset) {
       throw new ConcurrentFileModificationException(
           "offset commit rejected",
           FileserverFailureContext.forOffset(
@@ -150,6 +152,31 @@ public final class InMemoryUploadSessionStore implements UploadSessionStore {
             Optional.empty()));
   }
 
+  @Override
+  public boolean terminate(UploadId uploadId) {
+    UploadSession current = sessions.get(uploadId);
+    if (current == null || current.terminal()) {
+      return false;
+    }
+    sessions.put(uploadId, withState(current, current.leaseUntil(), true));
+    return true;
+  }
+
+  @Override
+  public boolean claimForCleanup(UploadId uploadId, Instant clockNow) {
+    UploadSession current = sessions.get(uploadId);
+    if (current == null || !current.terminal()) {
+      return false;
+    }
+    boolean leaseLapsed =
+        current.leaseUntil().isEmpty() || !current.leaseUntil().get().isAfter(clockNow);
+    if (!leaseLapsed) {
+      return false;
+    }
+    sessions.put(uploadId, withState(current, Optional.empty(), true));
+    return true;
+  }
+
   @Override
   public List<UploadSession> findExpired(Instant cutoff, int limit) {
     List<UploadSession> expired = new ArrayList<>();
@@ -188,6 +215,25 @@ public final class InMemoryUploadSessionStore implements UploadSessionStore {
         owner,
         token,
         leaseUntil,
+        current.terminal(),
+        current.version() + 1,
+        current.createdAt(),
+        now);
+  }
+
+  private UploadSession withState(
+      UploadSession current, Optional<Instant> leaseUntil, boolean terminal) {
+    return new UploadSession(
+        current.uploadId(),
+        current.fileId(),
+        current.protocol(),
+        current.expectedLength(),
+        current.committedOffset(),
+        current.expiresAt(),
+        leaseUntil.isEmpty() ? Optional.empty() : current.leaseOwner(),
+        leaseUntil.isEmpty() ? Optional.empty() : current.leaseToken(),
+        leaseUntil,
+        terminal,
         current.version() + 1,
         current.createdAt(),
         now);

src/application-core/src/test/java/dev/caskeleton/application/fileserver/upload/UploadApplicationServiceTest.java

@@ -5,6 +5,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import dev.caskeleton.application.fileserver.api.FileState;
 import dev.caskeleton.application.fileserver.api.UploadId;
+import dev.caskeleton.application.fileserver.api.error.ConcurrentFileModificationException;
 import dev.caskeleton.application.fileserver.api.error.FileAccessDeniedException;
 import dev.caskeleton.application.fileserver.api.error.FileTooLargeException;
 import dev.caskeleton.application.fileserver.api.error.StorageUnavailableException;
@@ -76,6 +77,33 @@ class UploadApplicationServiceTest {
     assertThat(cleanup.queued()).hasSize(1);
   }
 
+  /**
+   * Cancel closes the upload to writers, not just to readers.
+   *
+   * <p>It used to mark the record deleting and queue the staging cleanup and leave the session
+   * itself untouched, so a writer could still take a lease on the bytes the queued item was about
+   * to delete. "Cancelled" and "idle between appends" looked identical to every writer statement.
+   */
+  @Test
+  void cancelClosesTheUploadToFurtherWriters() {
+    UploadSessionView created = service.create(createRequest(), FileserverFixtures.context());
+    long version = sessions.find(created.uploadId()).orElseThrow().version();
+
+    service.cancel(created.uploadId(), FileserverFixtures.context());
+
+    assertThat(sessions.find(created.uploadId()).orElseThrow().terminal()).isTrue();
+    assertThatThrownBy(
+            () ->
+                sessions.acquireLease(
+                    created.uploadId(),
+                    "node-b",
+                    FileserverFixtures.NOW,
+                    java.time.Duration.ofSeconds(30),
+                    version))
+        .as("a lease granted here races the cleanup this cancel just queued")
+        .isInstanceOf(ConcurrentFileModificationException.class);
+  }
+
   @Test
   void createLeavesTheRecordUploadingWithAZeroOffset() {
     UploadSessionView created = service.create(createRequest(), FileserverFixtures.context());

src/application-core/src/test/java/dev/caskeleton/application/idempotency/IdempotencyExecutorV2Test.java

@@ -1,286 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.junit.jupiter.api.Test;
-
-class IdempotencyExecutorV2Test {
-
-  private static final String OWNER = "owner_token_1234567890";
-  private static final String OPERATION = "operation_token_12345";
-  private static final IdempotencyScope SCOPE =
-      IdempotencyScope.of("principal", "request", "create-worklog");
-  private static final RequestFingerprint FINGERPRINT = new RequestFingerprint("a".repeat(64));
-  private static final IdempotencyClaimAttempt ATTEMPT =
-      new IdempotencyClaimAttempt(OWNER, OPERATION);
-
-  @Test
-  void actionRunsOnlyAfterANewlyConfirmedStartAndThenCompletes() {
-    FakeStore store = new FakeStore();
-    IdempotencyOwner owner = new IdempotencyOwner(SCOPE, OWNER, 1);
-    store.claim =
-        new IdempotencyClaimOutcome.Acquired(owner, Instant.parse("2026-07-29T12:00:30Z"));
-    store.start = new IdempotencyStartOutcome(IdempotencyStartOutcome.Status.STARTED, null);
-    AtomicBoolean ran = new AtomicBoolean();
-
-    String result =
-        executor(store)
-            .execute(
-                SCOPE,
-                FINGERPRINT,
-                ATTEMPT,
-                () -> {
-                  ran.set(true);
-                  return new IdempotentAction.Outcome.Success<>("created");
-                },
-                codec());
-
-    assertThat(result).isEqualTo("created");
-    assertThat(ran).isTrue();
-    assertThat(store.completeCalls).isEqualTo(1);
-    assertThat(store.failedCalls).isZero();
-    assertThat(store.releaseCalls).isZero();
-  }
-
-  @Test
-  void exactSameOperationStartReplayOrInspectionConfirmsTheStartBeforeRunning() {
-    FakeStore store = new FakeStore();
-    IdempotencyOwner owner = new IdempotencyOwner(SCOPE, OWNER, 1);
-    store.claim = new IdempotencyClaimOutcome.Indeterminate(OPERATION);
-    store.inspection =
-        new IdempotencyInspection.ExecutingSameOperation(
-            owner, Instant.parse("2026-07-29T12:00:30Z"));
-    AtomicBoolean ran = new AtomicBoolean();
-
-    assertThat(
-            executor(store)
-                .execute(
-                    SCOPE,
-                    FINGERPRINT,
-                    ATTEMPT,
-                    () -> {
-                      ran.set(true);
-                      return new IdempotentAction.Outcome.Success<>("created");
-                    },
-                    codec()))
-        .isEqualTo("created");
-    assertThat(ran).isTrue();
-    assertThat(store.startCalls).isZero();
-
-    ran.set(false);
-    store.claim =
-        new IdempotencyClaimOutcome.ReplayedAcquire(owner, Instant.parse("2026-07-29T12:00:30Z"));
-    store.start =
-        new IdempotencyStartOutcome(
-            IdempotencyStartOutcome.Status.ALREADY_STARTED_SAME_OPERATION, null);
-    assertThat(
-            executor(store)
-                .execute(
-                    SCOPE,
-                    FINGERPRINT,
-                    ATTEMPT,
-                    () -> {
-                      ran.set(true);
-                      return new IdempotentAction.Outcome.Success<>("created-again");
-                    },
-                    codec()))
-        .isEqualTo("created-again");
-    assertThat(ran).isTrue();
-  }
-
-  @Test
-  void completionResponseLossIsInspectedWithoutRunningTheActionAgain() {
-    FakeStore store = startedStore();
-    store.complete =
-        new IdempotencyCompleteOutcome(IdempotencyCompleteOutcome.Status.INDETERMINATE, OPERATION);
-    store.inspectionAfterComplete =
-        new IdempotencyInspection.CompletedReplay(
-            new StoredResponse("created"), Instant.parse("2026-07-29T13:00:00Z"));
-    int[] actionCalls = {0};
-
-    assertThat(
-            executor(store)
-                .execute(
-                    SCOPE,
-                    FINGERPRINT,
-                    ATTEMPT,
-                    () -> {
-                      actionCalls[0]++;
-                      return new IdempotentAction.Outcome.Success<>("created");
-                    },
-                    codec()))
-        .isEqualTo("created");
-    assertThat(actionCalls[0]).isEqualTo(1);
-    assertThat(store.completeCalls).isEqualTo(1);
-  }
-
-  @Test
-  void unclassifiedExceptionsAndEffectUnknownAreNeverDiscardedAsNoEffect() {
-    FakeStore first = startedStore();
-    RuntimeException unclassified = new IllegalStateException("unknown effect");
-
-    assertThatThrownBy(
-            () ->
-                executor(first)
-                    .execute(
-                        SCOPE,
-                        FINGERPRINT,
-                        ATTEMPT,
-                        () -> {
-                          throw unclassified;
-                        },
-                        codec()))
-        .isSameAs(unclassified);
-    assertThat(first.lastDisposition)
-        .isEqualTo(IdempotencyFailureDisposition.ABANDONED_EFFECT_UNKNOWN);
-    assertThat(first.releaseCalls).isZero();
-
-    FakeStore store = startedStore();
-    RuntimeException classified = new IllegalArgumentException("provider response unknown");
-    FakeStore second = store;
-    assertThatThrownBy(
-            () ->
-                executor(second)
-                    .execute(
-                        SCOPE,
-                        FINGERPRINT,
-                        ATTEMPT,
-                        () -> new IdempotentAction.Outcome.EffectUnknown<>(classified),
-                        codec()))
-        .isSameAs(classified);
-    assertThat(second.lastDisposition)
-        .isEqualTo(IdempotencyFailureDisposition.ABANDONED_EFFECT_UNKNOWN);
-    assertThat(second.releaseCalls).isZero();
-  }
-
-  @Test
-  void explicitlyNoEffectFailureIsTheOnlyRetryableFailurePath() {
-    FakeStore store = startedStore();
-    RuntimeException failure = new IllegalArgumentException("validation");
-
-    assertThatThrownBy(
-            () ->
-                executor(store)
-                    .execute(
-                        SCOPE,
-                        FINGERPRINT,
-                        ATTEMPT,
-                        () -> new IdempotentAction.Outcome.RetryableNoEffect<>(failure),
-                        codec()))
-        .isSameAs(failure);
-    assertThat(store.lastDisposition).isEqualTo(IdempotencyFailureDisposition.RETRYABLE_NO_EFFECT);
-  }
-
-  private static FakeStore startedStore() {
-    FakeStore store = new FakeStore();
-    store.claim =
-        new IdempotencyClaimOutcome.Acquired(
-            new IdempotencyOwner(SCOPE, OWNER, 1), Instant.parse("2026-07-29T12:00:30Z"));
-    store.start = new IdempotencyStartOutcome(IdempotencyStartOutcome.Status.STARTED, null);
-    return store;
-  }
-
-  private static IdempotencyExecutorV2 executor(FakeStore store) {
-    return new IdempotencyExecutorV2(
-        store,
-        Duration.ofSeconds(30),
-        Duration.ofHours(1),
-        Duration.ofHours(1),
-        "json-v2",
-        "policy-v2");
-  }
-
-  private static IdempotentResponseCodec<String> codec() {
-    return new IdempotentResponseCodec<>() {
-      @Override
-      public String serialize(String result) {
-        return result;
-      }
-
-      @Override
-      public String deserialize(String payload) {
-        return payload;
-      }
-    };
-  }
-
-  private static final class FakeStore implements IdempotencyStorePortV2 {
-
-    private IdempotencyClaimOutcome claim;
-    private IdempotencyStartOutcome start;
-    private IdempotencyCompleteOutcome complete =
-        new IdempotencyCompleteOutcome(IdempotencyCompleteOutcome.Status.COMPLETED, null);
-    private IdempotencyInspection inspection = new IdempotencyInspection.Unavailable();
-    private IdempotencyInspection inspectionAfterComplete;
-    private int startCalls;
-    private int completeCalls;
-    private int failedCalls;
-    private int releaseCalls;
-    private IdempotencyFailureDisposition lastDisposition;
-
-    @Override
-    public IdempotencyClaimAttempt newClaimAttempt(String operationId) {
-      return new IdempotencyClaimAttempt(OWNER, operationId);
-    }
-
-    @Override
-    public IdempotencyClaimOutcome claim(IdempotencyClaimRequest request) {
-      return claim;
-    }
-
-    @Override
-    public IdempotencyStartOutcome markExecutionStarted(
-        IdempotencyOwner owner, String operationId) {
-      startCalls++;
-      return start;
-    }
-
-    @Override
-    public IdempotencyRenewOutcome renew(
-        IdempotencyOwner owner, Duration processingLeaseTtl, String operationId) {
-      throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public IdempotencyCompleteOutcome complete(
-        IdempotencyOwner owner, StoredResponse response, Duration replayTtl, String operationId) {
-      completeCalls++;
-      if (inspectionAfterComplete != null) {
-        inspection = inspectionAfterComplete;
-      }
-      return complete;
-    }
-
-    @Override
-    public IdempotencyFailOutcome markFailed(
-        IdempotencyOwner owner,
-        IdempotencyFailureDisposition disposition,
-        Duration retention,
-        String operationId) {
-      failedCalls++;
-      lastDisposition = disposition;
-      return new IdempotencyFailOutcome(
-          disposition == IdempotencyFailureDisposition.RETRYABLE_NO_EFFECT
-              ? IdempotencyFailOutcome.Status.MARKED_RETRYABLE
-              : IdempotencyFailOutcome.Status.MARKED_ABANDONED,
-          null);
-    }
-
-    @Override
-    public IdempotencyReleaseOutcome releaseBeforeExecution(
-        IdempotencyOwner owner, String operationId) {
-      releaseCalls++;
-      return new IdempotencyReleaseOutcome(
-          IdempotencyReleaseOutcome.Status.RELEASED_BEFORE_EXECUTION, null);
-    }
-
-    @Override
-    public IdempotencyInspection inspect(IdempotencyInspectionRequest request) {
-      return inspection;
-    }
-  }
-}

src/application-core/src/test/java/dev/caskeleton/application/idempotency/IdempotencyV2ContractSingularityTest.java

@@ -0,0 +1,72 @@
+package dev.caskeleton.application.idempotency;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The owner-safe request-replay contract exists in one package.
+ *
+ * <p>It existed in two. {@code dev.caskeleton.application.idempotency} declared a full {@code
+ * IdempotencyStorePortV2}, executor and outcome algebra, and {@code
+ * dev.caskeleton.application.idempotency.v2} declared another set under the same simple names.
+ * Every store — the Redis adapter and the PostgreSQL owner-safe one — implemented the {@code v2}
+ * contract, and nothing anywhere implemented the parent-package one.
+ *
+ * <p>That is not a harmless leftover. The composition root counts beans by type to decide whether a
+ * provider selection is complete, and a count taken against the unimplemented copy is a count that
+ * can never reach one: selecting a provider then failed with "ambiguous or incomplete" while every
+ * configured provider was present and correct. A second contract under the same names also means an
+ * import list decides which algebra a new adapter is written against, and the compiler cannot tell
+ * the author they picked the one no executor drives.
+ *
+ * <p>So the check is on the name, not on behaviour: the duplicate must stay deleted, because the
+ * failure it caused was invisible in every test that constructed its own store.
+ */
+class IdempotencyV2ContractSingularityTest {
+
+  private static final String V2_PACKAGE = "dev.caskeleton.application.idempotency.v2.";
+  private static final String PARENT_PACKAGE = "dev.caskeleton.application.idempotency.";
+
+  private static final String[] V2_TYPES = {
+    "IdempotencyStorePortV2",
+    "IdempotencyExecutorV2",
+    "IdempotencyClaimAttempt",
+    "IdempotencyClaimOutcome",
+    "IdempotencyClaimRequest",
+    "IdempotencyCompleteOutcome",
+    "IdempotencyFailOutcome",
+    "IdempotencyFailureDisposition",
+    "IdempotencyInspection",
+    "IdempotencyInspectionRequest",
+    "IdempotencyOwner",
+    "IdempotencyReleaseOutcome",
+    "IdempotencyRenewOutcome",
+    "IdempotencyStartOutcome"
+  };
+
+  @Test
+  @DisplayName("every owner-safe V2 type resolves in the v2 package")
+  void everyOwnerSafeTypeResolvesInTheV2Package() throws ClassNotFoundException {
+    for (String type : V2_TYPES) {
+      assertThat(Class.forName(V2_PACKAGE + type))
+          .as("%s is part of the contract the stores implement", type)
+          .isNotNull();
+    }
+  }
+
+  @Test
+  @DisplayName("no owner-safe V2 type has a second declaration in the parent package")
+  void noOwnerSafeTypeHasASecondDeclarationInTheParentPackage() {
+    for (String type : V2_TYPES) {
+      assertThatThrownBy(() -> Class.forName(PARENT_PACKAGE + type))
+          .as(
+              "%s must exist once; a second copy is what made the composition root count a bean"
+                  + " no provider could supply",
+              type)
+          .isInstanceOf(ClassNotFoundException.class);
+    }
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/idempotency/IdempotencyV2ContractTest.java

@@ -1,162 +0,0 @@
-package dev.caskeleton.application.idempotency;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-import java.time.Duration;
-import java.time.Instant;
-import org.junit.jupiter.api.Test;
-
-class IdempotencyV2ContractTest {
-
-  private static final String OWNER = "owner_token_1234567890";
-  private static final String OPERATION = "operation_token_12345";
-  private static final IdempotencyScope SCOPE =
-      IdempotencyScope.of("principal-digest", "request-key", "create-worklog");
-  private static final RequestFingerprint FINGERPRINT = new RequestFingerprint("a".repeat(64));
-
-  @Test
-  void claimSeparatesProcessingAndReplayRetentionAndCarriesAPreallocatedAttempt() {
-    IdempotencyClaimAttempt attempt = new IdempotencyClaimAttempt(OWNER, OPERATION);
-    IdempotencyClaimRequest request =
-        new IdempotencyClaimRequest(
-            SCOPE,
-            FINGERPRINT,
-            attempt,
-            Duration.ofSeconds(30),
-            Duration.ofHours(24),
-            "json-v2",
-            "request-replay-v2");
-
-    assertThat(request.processingLeaseTtl()).isEqualTo(Duration.ofSeconds(30));
-    assertThat(request.replayTtl()).isEqualTo(Duration.ofHours(24));
-    assertThat(request.recoveryRetention()).isEqualTo(Duration.ofHours(24));
-    assertThat(request.claimAttempt()).isSameAs(attempt);
-    assertThat(request.toString()).doesNotContain(OWNER).doesNotContain(OPERATION);
-  }
-
-  @Test
-  void ownerAndOperationTokensAreBoundedOpaqueAndRedacted() {
-    IdempotencyClaimAttempt attempt = new IdempotencyClaimAttempt(OWNER, OPERATION);
-    IdempotencyOwner owner = new IdempotencyOwner(SCOPE, OWNER, 3);
-
-    assertThat(attempt.toString()).doesNotContain(OWNER).doesNotContain(OPERATION);
-    assertThat(owner.toString()).doesNotContain(OWNER).doesNotContain("request-key");
-    assertThatThrownBy(() -> new IdempotencyClaimAttempt("short", OPERATION))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("ownerToken");
-    assertThatThrownBy(() -> new IdempotencyOwner(SCOPE, OWNER, 0))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("attempt");
-  }
-
-  @Test
-  void claimOutcomesCannotConfuseExpiredClaimedAndExpiredExecutingRecords() {
-    IdempotencyOwner owner = new IdempotencyOwner(SCOPE, OWNER, 2);
-    Instant leaseUntil = Instant.parse("2026-07-29T08:00:30Z");
-
-    IdempotencyClaimOutcome acquired = new IdempotencyClaimOutcome.Acquired(owner, leaseUntil);
-    IdempotencyClaimOutcome takeover =
-        new IdempotencyClaimOutcome.TakenOverClaimed(owner, leaseUntil);
-    IdempotencyClaimOutcome recovery = new IdempotencyClaimOutcome.RecoveryRequired(1);
-
-    assertThat(acquired).isInstanceOf(IdempotencyClaimOutcome.Acquired.class);
-    assertThat(takeover).isInstanceOf(IdempotencyClaimOutcome.TakenOverClaimed.class);
-    assertThat(recovery).isInstanceOf(IdempotencyClaimOutcome.RecoveryRequired.class);
-  }
-
-  @Test
-  void completedReplayAndInProgressHaveBoundedTypedPayloads() {
-    Instant replayUntil = Instant.parse("2026-07-30T08:00:00Z");
-    IdempotencyClaimOutcome completed =
-        new IdempotencyClaimOutcome.CompletedReplay(
-            new StoredResponse("{\"id\":\"42\"}"), replayUntil);
-    IdempotencyClaimOutcome inProgress =
-        new IdempotencyClaimOutcome.InProgress(Duration.ofMillis(250), 4);
-
-    assertThat(((IdempotencyClaimOutcome.CompletedReplay) completed).replayUntil())
-        .isEqualTo(replayUntil);
-    assertThat(((IdempotencyClaimOutcome.InProgress) inProgress).currentAttempt()).isEqualTo(4);
-    assertThatThrownBy(() -> new IdempotencyClaimOutcome.InProgress(Duration.ZERO, 1))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("retryAfter");
-  }
-
-  @Test
-  void everyOwnerMutationHasExplicitConflictUnknownAndUnavailableResults() {
-    assertThat(IdempotencyStartOutcome.operationConflict().status())
-        .isEqualTo(IdempotencyStartOutcome.Status.OPERATION_CONFLICT);
-    assertThat(IdempotencyRenewOutcome.indeterminate(OPERATION).status())
-        .isEqualTo(IdempotencyRenewOutcome.Status.INDETERMINATE);
-    assertThat(IdempotencyCompleteOutcome.responseConflict().status())
-        .isEqualTo(IdempotencyCompleteOutcome.Status.RESPONSE_CONFLICT);
-    assertThat(IdempotencyFailOutcome.unavailable().status())
-        .isEqualTo(IdempotencyFailOutcome.Status.UNAVAILABLE);
-    assertThat(IdempotencyReleaseOutcome.executionAlreadyStarted().status())
-        .isEqualTo(IdempotencyReleaseOutcome.Status.EXECUTION_ALREADY_STARTED);
-  }
-
-  @Test
-  void inspectionKeepsSameOperationRecoverySeparateFromOtherOwnerAndMismatch() {
-    IdempotencyOwner owner = new IdempotencyOwner(SCOPE, OWNER, 1);
-    Instant leaseUntil = Instant.parse("2026-07-29T08:00:30Z");
-    IdempotencyInspection sameOperation =
-        new IdempotencyInspection.ExecutingSameOperation(owner, leaseUntil);
-    IdempotencyInspection other = new IdempotencyInspection.InProgressOther(2);
-    IdempotencyInspection mismatch = new IdempotencyInspection.FingerprintMismatch();
-
-    assertThat(sameOperation).isInstanceOf(IdempotencyInspection.ExecutingSameOperation.class);
-    assertThat(other).isInstanceOf(IdempotencyInspection.InProgressOther.class);
-    assertThat(mismatch).isInstanceOf(IdempotencyInspection.FingerprintMismatch.class);
-  }
-
-  @Test
-  void ttlCodecAndPolicyFieldsAreFiniteAndBounded() {
-    assertThatThrownBy(
-            () ->
-                new IdempotencyClaimRequest(
-                    SCOPE,
-                    FINGERPRINT,
-                    new IdempotencyClaimAttempt(OWNER, OPERATION),
-                    Duration.ZERO,
-                    Duration.ofHours(1),
-                    "json-v2",
-                    "policy-v2"))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("processingLeaseTtl");
-    assertThatThrownBy(
-            () ->
-                new IdempotencyClaimRequest(
-                    SCOPE,
-                    FINGERPRINT,
-                    new IdempotencyClaimAttempt(OWNER, OPERATION),
-                    Duration.ofSeconds(30),
-                    Duration.ofDays(31),
-                    "json-v2",
-                    "policy-v2"))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("replayTtl");
-    assertThatThrownBy(
-            () ->
-                new IdempotencyClaimRequest(
-                    SCOPE,
-                    FINGERPRINT,
-                    new IdempotencyClaimAttempt(OWNER, OPERATION),
-                    Duration.ofSeconds(30),
-                    Duration.ofSeconds(30),
-                    "json-v2",
-                    "policy-v2"))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("recovery retention");
-  }
-
-  @Test
-  void requestFingerprintRejectsNonHexAndUppercaseRepresentations() {
-    assertThatThrownBy(() -> new RequestFingerprint("z".repeat(64)))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("lowercase");
-    assertThatThrownBy(() -> new RequestFingerprint("A".repeat(64)))
-        .isInstanceOf(IllegalArgumentException.class)
-        .hasMessageContaining("lowercase");
-  }
-}

src/application-core/src/test/java/dev/caskeleton/application/notification/platform/callback/CallbackFixtures.java

@@ -49,12 +49,27 @@ final class CallbackFixtures {
   }
 
   static ProviderEventRecord event(NormalizedEventType type, String nativeType) {
+    return event(type, nativeType, Optional.of("SM1"), Optional.empty());
+  }
+
+  /**
+   * A stored event as a sweep sees one: no raw provider request id, only the hash.
+   *
+   * @param providerRequestId the raw identifier, present only while the callback is in flight
+   * @param providerRequestIdHash the digest the row keeps
+   */
+  static ProviderEventRecord event(
+      NormalizedEventType type,
+      String nativeType,
+      Optional<String> providerRequestId,
+      Optional<ProviderRequestIdHash> providerRequestIdHash) {
     return new ProviderEventRecord(
         new ProviderEventRecordId(UUID.randomUUID()),
         PROFILE,
         new NormalizedProviderEvent(
-            type, nativeType, Optional.empty(), Optional.of("SM1"), Optional.of(NOW), Map.of()),
+            type, nativeType, Optional.empty(), providerRequestId, Optional.of(NOW), Map.of()),
         Optional.empty(),
+        providerRequestIdHash,
         ProviderEventSource.CALLBACK,
         true,
         NOW,

src/application-core/src/test/java/dev/caskeleton/application/notification/platform/callback/CallbackIngestionAtomicityTest.java

@@ -237,6 +237,11 @@ class CallbackIngestionAtomicityTest {
       return List.of();
     }
 
+    @Override
+    public boolean bindAttempt(ProviderEventRecordId eventId, DeliveryAttemptId attemptId) {
+      return false;
+    }
+
     @Override
     public List<ProviderEventRecord> eventsForAttempt(DeliveryAttemptId attemptId) {
       return List.of();

src/application-core/src/test/java/dev/caskeleton/application/notification/platform/callback/ProviderEventLateMatchTest.java

@@ -0,0 +1,292 @@
+package dev.caskeleton.application.notification.platform.callback;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import dev.caskeleton.application.notification.platform.api.DeliveryAttemptId;
+import dev.caskeleton.application.notification.platform.api.ProviderId;
+import dev.caskeleton.application.notification.platform.api.ProviderProfileId;
+import dev.caskeleton.application.notification.platform.api.delivery.DeliveryOutcome;
+import dev.caskeleton.application.notification.platform.api.delivery.EvidenceLevel;
+import dev.caskeleton.application.notification.platform.api.routing.Channel;
+import dev.caskeleton.application.notification.platform.observation.NotificationMetricsPort;
+import dev.caskeleton.application.transaction.TransactionPort;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * What a sweep can still match an already-stored event on.
+ *
+ * <p>A callback that beats its own attempt into the database is ordinary, and the ingestion path
+ * stores it unbound on purpose. Everything after that depended on a later pass finding the attempt
+ * — and the pass had nothing to look with. The raw provider request id is never stored, so a
+ * rehydrated event's {@code providerRequestId} is always empty, and an unbound event has no attempt
+ * id either. The sweep therefore looked at the same rows forever and resolved none of them.
+ */
+class ProviderEventLateMatchTest {
+
+  private static final ProviderId PROVIDER = new ProviderId("twilio");
+
+  private static final ProviderRequestIdHash HASH = new ProviderRequestIdHash("a".repeat(64));
+
+  private final DeliveryAttemptSnapshot attempt = CallbackFixtures.attempt(Channel.SMS, PROVIDER);
+
+  private final RecordingLedger ledger = new RecordingLedger();
+  private final RecordingResolver resolver = new RecordingResolver();
+  private final InMemoryProjections projections = new InMemoryProjections();
+
+  private final ProviderEventProjectionService projection =
+      new ProviderEventProjectionService(
+          ledger,
+          providerId -> Optional.of(new DeliveredProjector()),
+          resolver,
+          projections,
+          (ignoredAttempt, ignoredFacts) -> {},
+          new DirectTransactions(),
+          new DiscardingMetrics());
+
+  @Test
+  @DisplayName("a stored event with only its hash still finds its attempt")
+  void aStoredEventWithOnlyItsHashFindsItsAttempt() {
+    resolver.register(HASH, attempt);
+
+    Optional<ProjectionResult> result =
+        projection.project(
+            CallbackFixtures.event(
+                NormalizedEventType.DELIVERY_CONFIRMED,
+                "delivered",
+                Optional.empty(),
+                Optional.of(HASH)));
+
+    assertThat(result)
+        .as("the raw request id is not stored, so the hash is the only matcher a sweep has")
+        .isPresent();
+    assertThat(projections.saved()).containsKey(attempt.attemptId());
+  }
+
+  @Test
+  @DisplayName("the match is written back, so the next pass resolves by identity")
+  void theMatchIsWrittenBack() {
+    resolver.register(HASH, attempt);
+
+    ProviderEventRecord event =
+        CallbackFixtures.event(
+            NormalizedEventType.DELIVERY_CONFIRMED,
+            "delivered",
+            Optional.empty(),
+            Optional.of(HASH));
+    projection.project(event);
+
+    assertThat(ledger.bound())
+        .as("leaving attempt_id null makes every later pass hash its way back to the same attempt")
+        .containsExactly(Map.entry(event.id(), attempt.attemptId()));
+  }
+
+  @Test
+  @DisplayName("an event whose hash matches nothing stays unresolved rather than mismatched")
+  void anEventWhoseHashMatchesNothingStaysUnresolved() {
+    Optional<ProjectionResult> result =
+        projection.project(
+            CallbackFixtures.event(
+                NormalizedEventType.DELIVERY_CONFIRMED,
+                "delivered",
+                Optional.empty(),
+                Optional.of(HASH)));
+
+    assertThat(result).isEmpty();
+    assertThat(ledger.bound()).isEmpty();
+  }
+
+  @Test
+  @DisplayName("an event with no hash at all is left for an operator, not guessed at")
+  void anEventWithNoHashIsLeftAlone() {
+    resolver.register(HASH, attempt);
+
+    Optional<ProjectionResult> result =
+        projection.project(
+            CallbackFixtures.event(
+                NormalizedEventType.DELIVERY_CONFIRMED,
+                "delivered",
+                Optional.empty(),
+                Optional.empty()));
+
+    assertThat(result).isEmpty();
+    assertThat(ledger.bound()).isEmpty();
+  }
+
+  /** Records what the service asked the ledger to bind. */
+  private static final class RecordingLedger implements ProviderEventLedger {
+
+    private final List<Map.Entry<ProviderEventRecordId, DeliveryAttemptId>> bound =
+        new ArrayList<>();
+
+    private List<Map.Entry<ProviderEventRecordId, DeliveryAttemptId>> bound() {
+      return List.copyOf(bound);
+    }
+
+    @Override
+    public AppendEventResult append(VerifiedProviderEvent event) {
+      return new AppendEventResult(List.of(), List.of());
+    }
+
+    @Override
+    public AppendEventResult appendAll(List<VerifiedProviderEvent> events) {
+      return new AppendEventResult(List.of(), List.of());
+    }
+
+    @Override
+    public List<ProviderEventRecord> pendingProjection(int limit) {
+      return List.of();
+    }
+
+    @Override
+    public void markApplied(ProviderEventRecordId eventId, ProjectionResult result) {
+      // The status transition is the ledger's contract, not this one's.
+    }
+
+    @Override
+    public void markFailed(ProviderEventRecordId eventId, String errorCode) {
+      // Same.
+    }
+
+    @Override
+    public List<ProviderEventRecord> unmatched(int limit) {
+      return List.of();
+    }
+
+    @Override
+    public List<ProviderEventRecord> bindUnmatched(
+        ProviderProfileId providerProfileId,
+        String providerRequestId,
+        DeliveryAttemptId attemptId) {
+      return List.of();
+    }
+
+    @Override
+    public boolean bindAttempt(ProviderEventRecordId eventId, DeliveryAttemptId attemptId) {
+      bound.add(Map.entry(eventId, attemptId));
+      return true;
+    }
+
+    @Override
+    public List<ProviderEventRecord> eventsForAttempt(DeliveryAttemptId attemptId) {
+      return List.of();
+    }
+  }
+
+  /** Answers only for the hash it was given, and never for a raw identifier. */
+  private static final class RecordingResolver implements DeliveryAttemptResolverPort {
+
+    private final Map<ProviderRequestIdHash, DeliveryAttemptSnapshot> byHash =
+        new java.util.HashMap<>();
+
+    private void register(ProviderRequestIdHash hash, DeliveryAttemptSnapshot attempt) {
+      byHash.put(hash, attempt);
+    }
+
+    @Override
+    public Optional<DeliveryAttemptSnapshot> byAttemptId(DeliveryAttemptId attemptId) {
+      return Optional.empty();
+    }
+
+    @Override
+    public Optional<DeliveryAttemptSnapshot> byProviderRequestId(
+        ProviderProfileId profileId, String providerRequestId) {
+      return Optional.empty();
+    }
+
+    @Override
+    public Optional<DeliveryAttemptSnapshot> byProviderRequestIdHash(
+        ProviderProfileId profileId, ProviderRequestIdHash providerRequestIdHash) {
+      return Optional.ofNullable(byHash.get(providerRequestIdHash));
+    }
+  }
+
+  private static final class InMemoryProjections implements DeliveryProjectionStorePort {
+
+    private final Map<DeliveryAttemptId, DeliveryProjection> saved = new java.util.HashMap<>();
+
+    private Map<DeliveryAttemptId, DeliveryProjection> saved() {
+      return Map.copyOf(saved);
+    }
+
+    @Override
+    public DeliveryProjection load(DeliveryAttemptId attemptId) {
+      return saved.getOrDefault(attemptId, DeliveryProjection.accepted());
+    }
+
+    @Override
+    public void save(DeliveryAttemptId attemptId, DeliveryProjection projection) {
+      saved.put(attemptId, projection);
+    }
+
+    @Override
+    public boolean claimSuppressionSideEffect(DeliveryAttemptId attemptId) {
+      return false;
+    }
+  }
+
+  /** Marks the attempt delivered, which is enough to observe that projection ran. */
+  private static final class DeliveredProjector implements ProviderEventProjector {
+
+    @Override
+    public ProviderId providerId() {
+      return PROVIDER;
+    }
+
+    @Override
+    public ProjectionResult project(
+        DeliveryAttemptSnapshot attempt, ProviderEventRecord event, DeliveryProjection current) {
+      return ProjectionResult.applied(
+          current.withDelivery(DeliveryOutcome.DELIVERED, EvidenceLevel.DEVICE_DELIVERED),
+          "delivered");
+    }
+  }
+
+  /** Runs the action where it stands: this test is about matching, not about transactions. */
+  private static final class DirectTransactions implements TransactionPort {
+
+    @Override
+    public <T> T inWrite(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inRootWrite(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inRead(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inNew(Supplier<T> action) {
+      return action.get();
+    }
+  }
+
+  private static final class DiscardingMetrics implements NotificationMetricsPort {
+
+    @Override
+    public void increment(String metricName, Map<String, String> tags) {
+      // Nothing here asserts on metrics.
+    }
+
+    @Override
+    public void record(String metricName, Map<String, String> tags, Duration value) {
+      // Same.
+    }
+
+    @Override
+    public void gauge(String metricName, Map<String, String> tags, double value) {
+      // Same.
+    }
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/notification/platform/dispatch/PlatformFakes.java

@@ -25,11 +25,13 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
@@ -250,6 +252,33 @@ public final class PlatformFakes {
       requests.replace(moved);
       return moved;
     }
+
+    /**
+     * Leases this fake considers superseded. Add one and the fenced writes below stop matching, the
+     * way the database stops matching once another worker has claimed the job.
+     */
+    public final Set<RecipientLease> supersededLeases = new HashSet<>();
+
+    @Override
+    public Optional<RecipientDeliveryRecord> saveHeldBy(
+        RecipientDeliveryRecord record, RecipientLease lease) {
+      if (supersededLeases.contains(lease)) {
+        return Optional.empty();
+      }
+      return Optional.of(save(record));
+    }
+
+    @Override
+    public Optional<RecipientDeliveryRecord> transitionHeldBy(
+        RecipientDeliveryId id,
+        RecipientDeliveryState state,
+        Optional<Instant> nextDispatchAt,
+        RecipientLease lease) {
+      if (supersededLeases.contains(lease)) {
+        return Optional.empty();
+      }
+      return Optional.of(transition(id, state, nextDispatchAt));
+    }
   }
 
   /** Attempt store that keeps insertion order, which is what attempt numbering depends on. */

src/application-core/src/test/java/dev/caskeleton/application/outbox/OutcomeAwareRelayTest.java

@@ -0,0 +1,273 @@
+package dev.caskeleton.application.outbox;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import dev.caskeleton.application.transaction.TransactionPort;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.random.RandomGenerator;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * What the relay does with an answer the throwing port could not give.
+ *
+ * <p>{@code publishForOutcome} existed, no adapter overrode it and nothing called it: the relay
+ * called the throwing {@code publish}, so every outcome collapsed into "returned" or "threw". A
+ * broker that accepted a frame and never confirmed it was recorded exactly like one that refused
+ * the message outright — and a refusal that can never succeed spent the whole retry budget before
+ * being parked, which is the retry storm that makes an incident harder to read.
+ */
+class OutcomeAwareRelayTest {
+
+  private static final Instant NOW = Instant.parse("2026-08-15T12:00:00Z");
+
+  private static final Duration IN_FLIGHT_TIMEOUT = Duration.ofMinutes(5);
+
+  private OutcomeStore store;
+
+  private OutcomeReporter reporter;
+
+  private PublishPendingOutboxEventsUseCase useCase;
+
+  private OutcomePort publishPort;
+
+  @BeforeEach
+  void setUp() {
+    store = new OutcomeStore();
+    reporter = new OutcomeReporter();
+    publishPort = new OutcomePort();
+    useCase =
+        new PublishPendingOutboxEventsUseCase(
+            store,
+            publishPort,
+            reporter,
+            new DirectTransactions(),
+            new OutboxBackoffPolicy(new NoJitter()),
+            Clock.fixed(NOW, ZoneOffset.UTC),
+            10,
+            IN_FLIGHT_TIMEOUT);
+  }
+
+  @Test
+  @DisplayName("a confirmed publish still marks the row published")
+  void aConfirmedPublishMarksTheRowPublished() {
+    store.claimable.add(event("evt-1", 1));
+    publishPort.answer = OutboxPublishOutcome.CONFIRMED;
+
+    OutboxRelayResult result = useCase.handle(PublishPendingOutboxEventsCommand.INSTANCE);
+
+    assertThat(result.outcomes())
+        .extracting(OutboxRelayResult.EventOutcome::outcome)
+        .containsExactly(OutboxRelayResult.Outcome.PUBLISHED);
+    assertThat(store.published).containsExactly("evt-1");
+  }
+
+  @Test
+  @DisplayName("an ambiguous publish stays claimable rather than being treated as a definite loss")
+  void anAmbiguousPublishStaysClaimable() {
+    store.claimable.add(event("evt-2", 1));
+    publishPort.answer = OutboxPublishOutcome.AMBIGUOUS;
+
+    OutboxRelayResult result = useCase.handle(PublishPendingOutboxEventsCommand.INSTANCE);
+
+    assertThat(result.outcomes())
+        .extracting(OutboxRelayResult.EventOutcome::outcome)
+        .containsExactly(OutboxRelayResult.Outcome.FAILED);
+    assertThat(store.failed).containsKey("evt-2");
+    assertThat(store.dead).as("parking it loses a message that may have been delivered").isEmpty();
+    assertThat(reporter.reports)
+        .singleElement()
+        .satisfies(
+            report ->
+                assertThat(report.causeType())
+                    .as("the report names the state, so duplicates are seen coming")
+                    .isEqualTo(OutboxPublishAmbiguousException.class.getName()));
+  }
+
+  @Test
+  @DisplayName("a refusal before transmission is dead-lettered without spending the retry budget")
+  void aRefusalBeforeTransmissionIsDeadLettered() {
+    store.claimable.add(event("evt-3", 1));
+    publishPort.answer = OutboxPublishOutcome.REJECTED_BEFORE_SEND;
+
+    OutboxRelayResult result = useCase.handle(PublishPendingOutboxEventsCommand.INSTANCE);
+
+    assertThat(result.outcomes())
+        .extracting(OutboxRelayResult.EventOutcome::outcome)
+        .as("the same payload will be refused the same way on every attempt")
+        .containsExactly(OutboxRelayResult.Outcome.DEAD);
+    assertThat(store.dead).containsExactly("evt-3");
+    assertThat(store.failed).isEmpty();
+    assertThat(reporter.reports)
+        .singleElement()
+        .satisfies(
+            report ->
+                assertThat(report.causeType())
+                    .isEqualTo(OutboxPublishRefusedException.class.getName()));
+  }
+
+  @Test
+  @DisplayName("a broker-side refusal is also definite, and says which side refused")
+  void aBrokerSideRefusalIsAlsoDefinite() {
+    store.claimable.add(event("evt-4", 1));
+    publishPort.answer = OutboxPublishOutcome.REJECTED_AFTER_BROKER;
+
+    useCase.handle(PublishPendingOutboxEventsCommand.INSTANCE);
+
+    assertThat(store.dead).containsExactly("evt-4");
+    assertThat(reporter.reports.get(0).causeType())
+        .isEqualTo(OutboxPublishRefusedException.class.getName());
+    assertThat(new OutboxPublishRefusedException(OutboxPublishOutcome.REJECTED_AFTER_BROKER))
+        .as("which side refused is carried on the cause the report was built from")
+        .extracting(OutboxPublishRefusedException::outcome)
+        .isEqualTo(OutboxPublishOutcome.REJECTED_AFTER_BROKER);
+  }
+
+  @Test
+  @DisplayName("an adapter that only throws keeps the behaviour it always had")
+  void aThrowingAdapterIsUnchanged() {
+    store.claimable.add(event("evt-5", 1));
+    publishPort.failure = new IllegalStateException("the broker connection dropped");
+
+    OutboxRelayResult result = useCase.handle(PublishPendingOutboxEventsCommand.INSTANCE);
+
+    assertThat(result.outcomes())
+        .extracting(OutboxRelayResult.EventOutcome::outcome)
+        .containsExactly(OutboxRelayResult.Outcome.FAILED);
+    assertThat(store.failed).containsKey("evt-5");
+  }
+
+  private static OutboxEvent event(String eventId, int attemptCount) {
+    return new OutboxEvent(
+        eventId,
+        "UserCreated",
+        "agg-1",
+        "{\"data\": \"test\"}",
+        NOW.minusSeconds(60),
+        "corr-" + eventId,
+        "ikey-" + eventId,
+        OutboxEventStatus.IN_FLIGHT,
+        attemptCount);
+  }
+
+  /** A port that answers with an outcome, or throws when the test asks it to. */
+  private static final class OutcomePort implements OutboxMessagePublishPort {
+
+    private OutboxPublishOutcome answer = OutboxPublishOutcome.CONFIRMED;
+
+    private RuntimeException failure;
+
+    @Override
+    public void publish(OutboxEvent event) {
+      if (failure != null) {
+        throw failure;
+      }
+    }
+
+    @Override
+    public OutboxPublishOutcome publishForOutcome(OutboxEvent event) {
+      if (failure != null) {
+        throw failure;
+      }
+      return answer;
+    }
+  }
+
+  /** Records the transitions the relay drove. */
+  private static final class OutcomeStore implements OutboxStorePort {
+
+    private final List<OutboxEvent> claimable = new ArrayList<>();
+
+    private final List<String> published = new ArrayList<>();
+
+    private final java.util.Map<String, Instant> failed = new java.util.LinkedHashMap<>();
+
+    private final List<String> dead = new ArrayList<>();
+
+    @Override
+    public List<OutboxEvent> claimBatch(int batchSize, Instant now, Duration inFlightTimeout) {
+      List<OutboxEvent> claimed = List.copyOf(claimable);
+      claimable.clear();
+      return claimed;
+    }
+
+    @Override
+    public void markPublished(String eventId) {
+      published.add(eventId);
+    }
+
+    @Override
+    public void markFailed(String eventId, Instant nextAttemptAt) {
+      failed.put(eventId, nextAttemptAt);
+    }
+
+    @Override
+    public void markDead(String eventId) {
+      dead.add(eventId);
+    }
+
+    @Override
+    public java.util.Map<OutboxEventStatus, Long> countByStatus() {
+      return java.util.Map.of();
+    }
+
+    @Override
+    public java.util.Map<String, Long> oldestUnpublishedAgeSecondsByEventType(Instant now) {
+      return java.util.Map.of();
+    }
+  }
+
+  private static final class OutcomeReporter implements OutboxRelayFailureReportPort {
+
+    private final List<OutboxRelayFailureReport> reports = new ArrayList<>();
+
+    @Override
+    public void report(OutboxRelayFailureReport report) {
+      reports.add(report);
+    }
+  }
+
+  private static final class DirectTransactions implements TransactionPort {
+
+    @Override
+    public <T> T inWrite(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inRootWrite(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inRead(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inNew(Supplier<T> action) {
+      return action.get();
+    }
+  }
+
+  /** Deterministic backoff: the window is not what these cases are about. */
+  private static final class NoJitter implements RandomGenerator {
+
+    @Override
+    public long nextLong() {
+      return 0L;
+    }
+
+    @Override
+    public double nextDouble() {
+      return 0.0;
+    }
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/security/ObjectAccessPolicyTest.java

@@ -0,0 +1,93 @@
+package dev.caskeleton.application.security;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.RecordComponent;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The object-access rule is the application's to own, in terms no transport supplies.
+ *
+ * <p>The contract used to live in the inbound GraphQL adapter and take that adapter's request
+ * context, which meant no code in this module could implement it: doing so would have required
+ * application-core to depend on a transport, and every persistence adapter behind the use case with
+ * it. The neutrality case below is the one that keeps it that way — a signature that names an
+ * adapter type is the whole defect returning.
+ */
+class ObjectAccessPolicyTest {
+
+  @Test
+  void theContractNamesNoTransportType() {
+    List<String> transportTypes = new ArrayList<>();
+    for (Class<?> contract :
+        List.of(ObjectAccessPolicy.class, ObjectAccessRequest.class, ObjectAccessDecision.class)) {
+      for (Method method : contract.getDeclaredMethods()) {
+        collectAdapterTypes(method.getReturnType(), transportTypes);
+        for (Class<?> parameter : method.getParameterTypes()) {
+          collectAdapterTypes(parameter, transportTypes);
+        }
+      }
+      RecordComponent[] components = contract.getRecordComponents();
+      if (components != null) {
+        for (RecordComponent component : components) {
+          collectAdapterTypes(component.getType(), transportTypes);
+        }
+      }
+    }
+
+    assertThat(transportTypes)
+        .as("a contract that names an adapter type can only be implemented by that adapter")
+        .isEmpty();
+  }
+
+  @Test
+  void theBatchDefaultAnswersEveryRequestedIdInTheRequestedOrder() {
+    List<String> asked = new ArrayList<>();
+    ObjectAccessPolicy policy =
+        request -> {
+          asked.add(request.objectId());
+          return "order-2".equals(request.objectId())
+              ? ObjectAccessDecision.deny("OBJECT_NOT_AUTHORIZED")
+              : ObjectAccessDecision.allow();
+        };
+
+    Map<String, ObjectAccessDecision> decisions =
+        policy.decideAll("actor-1", "tenant-a", "Order", List.of("order-1", "order-2", "order-3"));
+
+    assertThat(asked).containsExactly("order-1", "order-2", "order-3");
+    assertThat(decisions).containsOnlyKeys("order-1", "order-2", "order-3");
+    assertThat(decisions.get("order-2").allowed()).isFalse();
+    assertThat(decisions.get("order-1").allowed()).isTrue();
+  }
+
+  @Test
+  void anUnauthenticatedCallerIsRepresentedByAnAbsentActorRatherThanABlankOne() {
+    ObjectAccessRequest request = new ObjectAccessRequest(null, "tenant-a", "Order", "order-1");
+
+    assertThat(request.actor()).isEmpty();
+    assertThatThrownBy(() -> new ObjectAccessRequest("actor-1", "tenant-a", "Order", " "))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void aDenialWithoutAStableCodeIsRejected() {
+    assertThatThrownBy(() -> new ObjectAccessDecision(false, null, false))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new ObjectAccessDecision(true, "WHY", false))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThat(ObjectAccessDecision.denyHidingExistence("OBJECT_NOT_FOUND").hideExistence())
+        .isTrue();
+  }
+
+  private static void collectAdapterTypes(Class<?> type, List<String> found) {
+    Class<?> element = type.isArray() ? type.getComponentType() : type;
+    if (element.getName().startsWith("dev.caskeleton.adapter")) {
+      found.add(element.getName());
+    }
+  }
+}