feat: add JPA production capability

src/application-core/README.md

@@ -256,6 +256,21 @@ application 계층이 in-flight 대기·replay **정책** 을 소유하고, 저
 - retryable: mismatch / in-flight 모두 `false`. 클라이언트는 body 를 고치거나 결과를 polling 해야지
   단순 재시도를 하면 안 된다.
 
+### Owner-safe idempotency V2
+
+`idempotency.v2`는 V1의 scope-only `tryBegin/complete/discard`를 대체하는 additive contract다.
+provider가 JPA인지 Redis인지와 무관하게 claim에는 secure owner token과 stable operation ID가
+필요하고, 모든 mutation은 owner/attempt/claim-operation/state-revision을 검증한다.
+
+- processing lease와 completed replay TTL을 분리한다.
+- expired `CLAIMED`만 takeover하고 expired `EXECUTING`은 `RECOVERY_REQUIRED`로 닫는다.
+- complete/fail/release는 operation ID와 result digest가 같은 재호출만 prior result로 replay한다.
+- `SAME_STORE_TRANSACTIONAL` JPA profile의 response는 8 KiB 이하 inline 값만 지원한다.
+- raw principal/client key는 versioned HMAC scope digest로 바꾼 뒤 adapter에 전달한다.
+
+V1은 rolling migration compatibility를 위해 유지된다. 새 reliability profile이 V2 claim과 V1
+scope-only mutation을 섞는 것은 금지한다.
+
 ---
 
 ## 트랜잭셔널 아웃박스 릴레이 (outbox)
@@ -383,6 +398,26 @@ claim → 트랜잭션 밖에서 발행 → at-least-once 보장.
   adapter에 전달하는 framework-free outbound contract. 구조화 ERROR 필드와 runbook 렌더링은
   messaging adapter가 소유한다.
 
+### Immutable outbox/polling delivery V2
+
+`outbox.v2`는 domain event intent와 delivery state를 분리한다.
+
+- `NewOutboxEventV2`의 aggregate version과 deterministic ordinal이 ordering authority다.
+- `OutboxAppendPortV2`는 caller의 primary write transaction에 참여하고 publication epoch와
+  authority를 DB control row에서 얻는다.
+- immutable event ID 충돌과 aggregate ordering tuple 충돌은 서로 다른 outcome이다.
+- `OutboxPollingDeliveryPortV2`는 publish 밖의 짧은 claim/completion transaction만 소유하고
+  owner/token/attempt/version/epoch CAS로 stale relay를 거절한다.
+- broker publish와 DB completion 사이 ACK 유실은 stable event ID의 duplicate publish를 만들 수
+  있으므로 exactly-once delivery로 표현하지 않는다.
+
+### Same-store inbox
+
+`inbox.InboxStorePort`는 broker redelivery를 DB business mutation과 같은 transaction에서
+deduplicate한다. `RECEIVED -> PROCESSING -> COMPLETED`가 기본이며 expired `PROCESSING`은 blind
+takeover하지 않는다. broker ACK는 transaction commit 이후 adapter 바깥에서만 수행하고 remote
+side effect는 outbox/workflow로 옮긴다.
+
 ---
 
 ## 분산 락 (lock)

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

@@ -0,0 +1,20 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/** Caller-retained identity for one claim send/retry sequence. */
+public record IdempotencyClaimAttempt(String ownerToken, OperationId operationId) {
+
+  private static final Pattern OWNER_TOKEN = Pattern.compile("[0-9a-f]{64}");
+
+  public IdempotencyClaimAttempt {
+    Objects.requireNonNull(ownerToken, "ownerToken");
+    Objects.requireNonNull(operationId, "operationId");
+    if (!OWNER_TOKEN.matcher(ownerToken).matches()) {
+      throw new IllegalArgumentException(
+          "owner token must be a 64-character lowercase hexadecimal secure-random value");
+    }
+  }
+}

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

@@ -0,0 +1,73 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.idempotency.StoredResponse;
+import dev.caskeleton.application.transaction.OperationId;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+
+/** Exhaustive result of an atomic owner-safe claim. */
+public sealed interface IdempotencyClaimOutcome {
+
+  record Acquired(IdempotencyOwner owner, Instant processingLeaseUntil)
+      implements IdempotencyClaimOutcome {
+    public Acquired {
+      Objects.requireNonNull(owner, "owner");
+      Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
+    }
+  }
+
+  record ReplayedAcquire(IdempotencyOwner owner, Instant processingLeaseUntil)
+      implements IdempotencyClaimOutcome {
+    public ReplayedAcquire {
+      Objects.requireNonNull(owner, "owner");
+      Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
+    }
+  }
+
+  record TakenOverClaimed(IdempotencyOwner owner, Instant processingLeaseUntil)
+      implements IdempotencyClaimOutcome {
+    public TakenOverClaimed {
+      Objects.requireNonNull(owner, "owner");
+      Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
+    }
+  }
+
+  record CompletedReplay(StoredResponse response, Instant replayUntil)
+      implements IdempotencyClaimOutcome {
+    public CompletedReplay {
+      Objects.requireNonNull(response, "response");
+      Objects.requireNonNull(replayUntil, "replayUntil");
+    }
+  }
+
+  record InProgress(Duration retryAfter, long currentAttempt) implements IdempotencyClaimOutcome {
+    public InProgress {
+      Objects.requireNonNull(retryAfter, "retryAfter");
+      if (retryAfter.isNegative() || currentAttempt < 1) {
+        throw new IllegalArgumentException(
+            "retry-after must be non-negative and current attempt must be positive");
+      }
+    }
+  }
+
+  record RecoveryRequired(long currentAttempt) implements IdempotencyClaimOutcome {
+    public RecoveryRequired {
+      if (currentAttempt < 1) {
+        throw new IllegalArgumentException("current attempt must be positive");
+      }
+    }
+  }
+
+  record FingerprintMismatch() implements IdempotencyClaimOutcome {}
+
+  record OwnerOperationConflict() implements IdempotencyClaimOutcome {}
+
+  record Indeterminate(OperationId operationId) implements IdempotencyClaimOutcome {
+    public Indeterminate {
+      Objects.requireNonNull(operationId, "operationId");
+    }
+  }
+
+  record Unavailable() implements IdempotencyClaimOutcome {}
+}

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

@@ -0,0 +1,41 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.idempotency.RequestFingerprint;
+import java.time.Duration;
+import java.util.Objects;
+
+/** Complete provider-neutral intent for one atomic owner-safe claim. */
+public record IdempotencyClaimRequest(
+    IdempotencyScopeDigest scope,
+    RequestFingerprint requestFingerprint,
+    IdempotencyClaimAttempt claimAttempt,
+    Duration processingLeaseTtl,
+    Duration replayTtl,
+    String responseCodecId,
+    int policyRevision) {
+
+  private static final Duration MAXIMUM_PROCESSING_LEASE = Duration.ofHours(1);
+  private static final Duration MAXIMUM_REPLAY_TTL = Duration.ofDays(30);
+
+  public IdempotencyClaimRequest {
+    Objects.requireNonNull(scope, "scope");
+    Objects.requireNonNull(requestFingerprint, "requestFingerprint");
+    Objects.requireNonNull(claimAttempt, "claimAttempt");
+    requirePositiveBounded("processing lease TTL", processingLeaseTtl, MAXIMUM_PROCESSING_LEASE);
+    requirePositiveBounded("replay TTL", replayTtl, MAXIMUM_REPLAY_TTL);
+    Objects.requireNonNull(responseCodecId, "responseCodecId");
+    if (responseCodecId.isBlank() || responseCodecId.length() > 64) {
+      throw new IllegalArgumentException("response codec ID must contain 1-64 characters");
+    }
+    if (policyRevision < 1) {
+      throw new IllegalArgumentException("policy revision must be positive");
+    }
+  }
+
+  private static void requirePositiveBounded(String name, Duration value, Duration maximum) {
+    Objects.requireNonNull(value, name);
+    if (value.isZero() || value.isNegative() || value.compareTo(maximum) > 0) {
+      throw new IllegalArgumentException(name + " must be positive and at most " + maximum);
+    }
+  }
+}

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

@@ -0,0 +1,14 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Outcomes of an owner-safe EXECUTING to COMPLETED transition. */
+public enum IdempotencyCompleteOutcome {
+  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/v2/IdempotencyFailOutcome.java

@@ -0,0 +1,14 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Outcomes of an owner-safe failure transition. */
+public enum IdempotencyFailOutcome {
+  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/v2/IdempotencyFailureDisposition.java

@@ -0,0 +1,7 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Whether a failed action is proven retryable or requires explicit effect reconciliation. */
+public enum IdempotencyFailureDisposition {
+  NO_EFFECT_RETRYABLE,
+  EFFECT_UNKNOWN_ABANDONED
+}

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

@@ -0,0 +1,28 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.idempotency.StoredResponse;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Inspection classification with only the data meaningful for that classification. */
+public record IdempotencyInspection(
+    IdempotencyInspectionOutcome outcome,
+    Optional<IdempotencyOwner> owner,
+    Optional<Instant> processingLeaseUntil,
+    Optional<StoredResponse> response,
+    Optional<Instant> replayUntil) {
+
+  public IdempotencyInspection {
+    Objects.requireNonNull(outcome, "outcome");
+    Objects.requireNonNull(owner, "owner");
+    Objects.requireNonNull(processingLeaseUntil, "processingLeaseUntil");
+    Objects.requireNonNull(response, "response");
+    Objects.requireNonNull(replayUntil, "replayUntil");
+  }
+
+  public static IdempotencyInspection outcome(IdempotencyInspectionOutcome outcome) {
+    return new IdempotencyInspection(
+        outcome, Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/v2/IdempotencyInspectionOutcome.java

@@ -0,0 +1,15 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Recovery-safe classifications returned by {@link IdempotencyStorePortV2#inspect}. */
+public enum IdempotencyInspectionOutcome {
+  ABSENT,
+  CLAIMED_SAME_OPERATION,
+  EXECUTING_SAME_OPERATION,
+  COMPLETED_REPLAY,
+  IN_PROGRESS_OTHER,
+  FAILED_RETRYABLE,
+  ABANDONED,
+  FINGERPRINT_MISMATCH,
+  OPERATION_CONFLICT,
+  UNAVAILABLE
+}

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

@@ -0,0 +1,17 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.idempotency.RequestFingerprint;
+import java.util.Objects;
+
+/** Read-only recovery request after a claim or transition response was lost. */
+public record IdempotencyInspectionRequest(
+    IdempotencyScopeDigest scope,
+    RequestFingerprint requestFingerprint,
+    IdempotencyClaimAttempt claimAttempt) {
+
+  public IdempotencyInspectionRequest {
+    Objects.requireNonNull(scope, "scope");
+    Objects.requireNonNull(requestFingerprint, "requestFingerprint");
+    Objects.requireNonNull(claimAttempt, "claimAttempt");
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/v2/IdempotencyMutationResult.java

@@ -0,0 +1,35 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Predicate;
+
+/**
+ * Operation-specific typed outcome plus a replacement owner handle when ownership remains valid.
+ */
+public final class IdempotencyMutationResult<O extends Enum<O>> {
+
+  private final O outcome;
+  private final IdempotencyOwner owner;
+
+  public IdempotencyMutationResult(O outcome, IdempotencyOwner owner, Predicate<O> carriesOwner) {
+    this.outcome = Objects.requireNonNull(outcome, "outcome");
+    Objects.requireNonNull(carriesOwner, "carriesOwner");
+    boolean expectedOwner = carriesOwner.test(outcome);
+    if (expectedOwner && owner == null) {
+      throw new IllegalArgumentException(outcome + " must carry the current owner handle");
+    }
+    if (!expectedOwner && owner != null) {
+      throw new IllegalArgumentException(outcome + " must not carry an owner handle");
+    }
+    this.owner = owner;
+  }
+
+  public O outcome() {
+    return outcome;
+  }
+
+  public Optional<IdempotencyOwner> owner() {
+    return Optional.ofNullable(owner);
+  }
+}

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

@@ -0,0 +1,37 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.util.Objects;
+
+/**
+ * Owner handle carrying the full optimistic CAS tuple.
+ *
+ * <p>Every successful state-changing operation returns a replacement handle with the incremented
+ * state revision. A stale handle is never silently accepted.
+ */
+public record IdempotencyOwner(
+    IdempotencyScopeDigest scope,
+    String ownerToken,
+    long attempt,
+    long stateRevision,
+    OperationId claimOperationId) {
+
+  public IdempotencyOwner {
+    Objects.requireNonNull(scope, "scope");
+    Objects.requireNonNull(ownerToken, "ownerToken");
+    Objects.requireNonNull(claimOperationId, "claimOperationId");
+    if (ownerToken.isBlank() || ownerToken.length() > 128) {
+      throw new IllegalArgumentException("owner token must contain 1-128 characters");
+    }
+    if (attempt < 1) {
+      throw new IllegalArgumentException("attempt must be positive");
+    }
+    if (stateRevision < 0) {
+      throw new IllegalArgumentException("state revision must be non-negative");
+    }
+  }
+
+  public IdempotencyOwner withStateRevision(long nextRevision) {
+    return new IdempotencyOwner(scope, ownerToken, attempt, nextRevision, claimOperationId);
+  }
+}

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

@@ -0,0 +1,13 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Outcomes of releasing a claim only while business execution has not started. */
+public enum IdempotencyReleaseOutcome {
+  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/v2/IdempotencyRenewOutcome.java

@@ -0,0 +1,23 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Outcomes of an owner-safe processing lease renewal. */
+public enum IdempotencyRenewOutcome {
+  RENEWED(true),
+  ALREADY_RENEWED_SAME_OPERATION(true),
+  ABSENT(false),
+  NOT_OWNER(false),
+  NOT_IN_PROGRESS(false),
+  OPERATION_CONFLICT(false),
+  INDETERMINATE(false),
+  UNAVAILABLE(false);
+
+  private final boolean carriesOwner;
+
+  IdempotencyRenewOutcome(boolean carriesOwner) {
+    this.carriesOwner = carriesOwner;
+  }
+
+  public boolean carriesOwner() {
+    return carriesOwner;
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/v2/IdempotencyScopeDigest.java

@@ -0,0 +1,32 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * Provider-neutral, versioned HMAC digest of the canonical idempotency scope.
+ *
+ * <p>The raw client key and principal must be digested before this value is constructed. Database,
+ * Redis, logs, and metrics receive only this opaque value.
+ */
+public record IdempotencyScopeDigest(String digest, int keyDigestVersion, String operationCode) {
+
+  private static final Pattern LOWERCASE_SHA_256 = Pattern.compile("[0-9a-f]{64}");
+  private static final Pattern OPERATION_CODE = Pattern.compile("[A-Z][A-Z0-9_]{0,63}");
+
+  public IdempotencyScopeDigest {
+    Objects.requireNonNull(digest, "digest");
+    Objects.requireNonNull(operationCode, "operationCode");
+    if (!LOWERCASE_SHA_256.matcher(digest).matches()) {
+      throw new IllegalArgumentException(
+          "scope digest must be a 64-character lowercase hexadecimal HMAC-SHA-256 value");
+    }
+    if (keyDigestVersion < 1) {
+      throw new IllegalArgumentException("key digest version must be positive");
+    }
+    if (!OPERATION_CODE.matcher(operationCode).matches()) {
+      throw new IllegalArgumentException(
+          "operation code must be 1-64 uppercase ASCII letters, digits, or underscores");
+    }
+  }
+}

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

@@ -0,0 +1,23 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Outcomes of the CLAIMED to EXECUTING owner-safe transition. */
+public enum IdempotencyStartOutcome {
+  STARTED(true),
+  ALREADY_STARTED_SAME_OPERATION(true),
+  ABSENT(false),
+  NOT_OWNER(false),
+  NOT_CLAIMED(false),
+  OPERATION_CONFLICT(false),
+  INDETERMINATE(false),
+  UNAVAILABLE(false);
+
+  private final boolean carriesOwner;
+
+  IdempotencyStartOutcome(boolean carriesOwner) {
+    this.carriesOwner = carriesOwner;
+  }
+
+  public boolean carriesOwner() {
+    return carriesOwner;
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/idempotency/v2/IdempotencyState.java

@@ -0,0 +1,10 @@
+package dev.caskeleton.application.idempotency.v2;
+
+/** Owner-safe idempotency V2 state machine. */
+public enum IdempotencyState {
+  CLAIMED,
+  EXECUTING,
+  COMPLETED,
+  FAILED_RETRYABLE,
+  ABANDONED
+}

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

@@ -0,0 +1,37 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import dev.caskeleton.application.idempotency.StoredResponse;
+import dev.caskeleton.application.transaction.OperationId;
+import java.time.Duration;
+
+/**
+ * Provider-neutral owner-safe idempotency state machine.
+ *
+ * <p>V1 remains source-compatible during migration, but new reliability profiles must use this
+ * complete contract rather than combining V2 claim with scope-only V1 mutations.
+ */
+public interface IdempotencyStorePortV2 {
+
+  IdempotencyClaimAttempt newClaimAttempt(OperationId operationId);
+
+  IdempotencyClaimOutcome claim(IdempotencyClaimRequest request);
+
+  IdempotencyMutationResult<IdempotencyStartOutcome> markExecutionStarted(
+      IdempotencyOwner owner, OperationId operationId);
+
+  IdempotencyMutationResult<IdempotencyRenewOutcome> renew(
+      IdempotencyOwner owner, Duration processingLeaseTtl, OperationId operationId);
+
+  IdempotencyCompleteOutcome complete(
+      IdempotencyOwner owner, StoredResponse response, Duration replayTtl, OperationId operationId);
+
+  IdempotencyFailOutcome markFailed(
+      IdempotencyOwner owner,
+      IdempotencyFailureDisposition disposition,
+      Duration retention,
+      OperationId operationId);
+
+  IdempotencyReleaseOutcome releaseBeforeExecution(IdempotencyOwner owner, OperationId operationId);
+
+  IdempotencyInspection inspect(IdempotencyInspectionRequest request);
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxClaimAttempt.java

@@ -0,0 +1,20 @@
+package dev.caskeleton.application.inbox;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/** Caller-retained inbox claim identity used to replay a lost claim response. */
+public record InboxClaimAttempt(String ownerToken, OperationId operationId) {
+
+  private static final Pattern TOKEN = Pattern.compile("[0-9a-f]{64}");
+
+  public InboxClaimAttempt {
+    Objects.requireNonNull(ownerToken, "ownerToken");
+    Objects.requireNonNull(operationId, "operationId");
+    if (!TOKEN.matcher(ownerToken).matches()) {
+      throw new IllegalArgumentException(
+          "owner token must be a 64-character lowercase hexadecimal value");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxClaimOutcome.java

@@ -0,0 +1,40 @@
+package dev.caskeleton.application.inbox;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+
+/** Exhaustive atomic inbox claim classification. */
+public sealed interface InboxClaimOutcome {
+
+  record Acquired(InboxOwner owner, Instant leaseUntil) implements InboxClaimOutcome {
+    public Acquired {
+      Objects.requireNonNull(owner, "owner");
+      Objects.requireNonNull(leaseUntil, "leaseUntil");
+    }
+  }
+
+  record ReplayedAcquire(InboxOwner owner, Instant leaseUntil) implements InboxClaimOutcome {
+    public ReplayedAcquire {
+      Objects.requireNonNull(owner, "owner");
+      Objects.requireNonNull(leaseUntil, "leaseUntil");
+    }
+  }
+
+  record TakenOver(InboxOwner owner, Instant leaseUntil) implements InboxClaimOutcome {
+    public TakenOver {
+      Objects.requireNonNull(owner, "owner");
+      Objects.requireNonNull(leaseUntil, "leaseUntil");
+    }
+  }
+
+  record Completed() implements InboxClaimOutcome {}
+
+  record InProgress(Duration retryAfter, long attempt) implements InboxClaimOutcome {}
+
+  record RecoveryRequired(long attempt) implements InboxClaimOutcome {}
+
+  record IntentMismatch() implements InboxClaimOutcome {}
+
+  record OwnerOperationConflict() implements InboxClaimOutcome {}
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxClaimRequest.java

@@ -0,0 +1,35 @@
+package dev.caskeleton.application.inbox;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/** Atomic inbox claim request with separate processing and terminal retention windows. */
+public record InboxClaimRequest(
+    InboxScopeDigest scope,
+    String messageIntentDigest,
+    InboxClaimAttempt claimAttempt,
+    Duration processingLease,
+    Duration terminalRetention) {
+
+  private static final Pattern DIGEST = Pattern.compile("[0-9a-f]{64}");
+
+  public InboxClaimRequest {
+    Objects.requireNonNull(scope, "scope");
+    Objects.requireNonNull(messageIntentDigest, "messageIntentDigest");
+    Objects.requireNonNull(claimAttempt, "claimAttempt");
+    if (!DIGEST.matcher(messageIntentDigest).matches()) {
+      throw new IllegalArgumentException(
+          "message intent digest must be a 64-character lowercase hexadecimal value");
+    }
+    requirePositiveBounded("processing lease", processingLease, Duration.ofHours(1));
+    requirePositiveBounded("terminal retention", terminalRetention, Duration.ofDays(30));
+  }
+
+  private static void requirePositiveBounded(String name, Duration value, Duration maximum) {
+    Objects.requireNonNull(value, name);
+    if (value.isZero() || value.isNegative() || value.compareTo(maximum) > 0) {
+      throw new IllegalArgumentException(name + " must be positive and at most " + maximum);
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxOwner.java

@@ -0,0 +1,32 @@
+package dev.caskeleton.application.inbox;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.util.Objects;
+
+/** Full owner/attempt/claim-operation/state-revision CAS tuple for one inbox message. */
+public record InboxOwner(
+    InboxScopeDigest scope,
+    String ownerToken,
+    long attempt,
+    long stateRevision,
+    OperationId claimOperationId) {
+
+  public InboxOwner {
+    Objects.requireNonNull(scope, "scope");
+    Objects.requireNonNull(ownerToken, "ownerToken");
+    Objects.requireNonNull(claimOperationId, "claimOperationId");
+    if (ownerToken.isBlank() || ownerToken.length() > 128) {
+      throw new IllegalArgumentException("owner token must contain 1-128 characters");
+    }
+    if (attempt < 1) {
+      throw new IllegalArgumentException("attempt must be positive");
+    }
+    if (stateRevision < 0) {
+      throw new IllegalArgumentException("state revision must be non-negative");
+    }
+  }
+
+  public InboxOwner withStateRevision(long revision) {
+    return new InboxOwner(scope, ownerToken, attempt, revision, claimOperationId);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxOwnerTransition.java

@@ -0,0 +1,20 @@
+package dev.caskeleton.application.inbox;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/** Transition outcome and updated owner handle while ownership remains live. */
+public record InboxOwnerTransition(InboxTransitionOutcome outcome, Optional<InboxOwner> owner) {
+
+  public InboxOwnerTransition {
+    Objects.requireNonNull(outcome, "outcome");
+    Objects.requireNonNull(owner, "owner");
+    boolean mustCarryOwner = outcome == InboxTransitionOutcome.PROCESSING_STARTED;
+    if (mustCarryOwner != owner.isPresent()) {
+      throw new IllegalArgumentException(
+          mustCarryOwner
+              ? "PROCESSING_STARTED must carry an updated owner"
+              : outcome + " must not carry an owner");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxScopeDigest.java

@@ -0,0 +1,18 @@
+package dev.caskeleton.application.inbox;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/** Versioned canonical digest of consumer-group, handler, tenant, and message ID scope. */
+public record InboxScopeDigest(String value) {
+
+  private static final Pattern LOWERCASE_SHA_256 = Pattern.compile("[0-9a-f]{64}");
+
+  public InboxScopeDigest {
+    Objects.requireNonNull(value, "value");
+    if (!LOWERCASE_SHA_256.matcher(value).matches()) {
+      throw new IllegalArgumentException(
+          "inbox scope must be a 64-character lowercase hexadecimal digest");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxState.java

@@ -0,0 +1,10 @@
+package dev.caskeleton.application.inbox;
+
+/** Same-store inbox lifecycle. */
+public enum InboxState {
+  RECEIVED,
+  PROCESSING,
+  COMPLETED,
+  RETRYABLE,
+  DEAD
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxStorePort.java

@@ -0,0 +1,23 @@
+package dev.caskeleton.application.inbox;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.time.Duration;
+
+/**
+ * Owner-safe same-store inbox state machine. Broker ACK must happen only after transaction commit.
+ */
+public interface InboxStorePort {
+
+  InboxClaimAttempt newClaimAttempt(OperationId operationId);
+
+  InboxClaimOutcome claim(InboxClaimRequest request);
+
+  InboxOwnerTransition markProcessing(InboxOwner owner, OperationId operationId);
+
+  InboxTransitionOutcome complete(InboxOwner owner, OperationId operationId);
+
+  InboxTransitionOutcome markRetryable(
+      InboxOwner owner, Duration retention, OperationId operationId);
+
+  InboxTransitionOutcome markDead(InboxOwner owner, Duration retention, OperationId operationId);
+}

src/application-core/src/main/java/dev/caskeleton/application/inbox/InboxTransitionOutcome.java

@@ -0,0 +1,15 @@
+package dev.caskeleton.application.inbox;
+
+/** Owner-safe inbox transition classification. */
+public enum InboxTransitionOutcome {
+  PROCESSING_STARTED,
+  COMPLETED,
+  RETRYABLE,
+  DEAD,
+  ALREADY_APPLIED_SAME_OPERATION,
+  RESULT_CONFLICT,
+  ABSENT,
+  NOT_OWNER,
+  INVALID_STATE,
+  STALE_REVISION
+}

src/application-core/src/main/java/dev/caskeleton/application/outbound/CallBudget.java

@@ -27,7 +27,12 @@ public record CallBudget(long monotonicDeadlineNanos) {
       throw new IllegalArgumentException(
           "call budget duration exceeds the supported range", exception);
     }
-    return new CallBudget(monotonicNowNanos + durationNanos);
+    try {
+      return new CallBudget(Math.addExact(monotonicNowNanos, durationNanos));
+    } catch (ArithmeticException exception) {
+      throw new IllegalArgumentException(
+          "call budget deadline exceeds the monotonic range", exception);
+    }
   }
 
   public long remainingNanosAt(long monotonicNowNanos) {

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/ClaimedOutboxDelivery.java

@@ -0,0 +1,35 @@
+package dev.caskeleton.application.outbox.v2;
+
+import java.time.Instant;
+import java.util.Objects;
+
+/** Immutable message envelope plus the mutable delivery owner handle. */
+public record ClaimedOutboxDelivery(
+    OutboxDeliveryOwner owner,
+    String eventType,
+    int eventSchema,
+    String aggregateType,
+    String aggregateId,
+    long aggregateVersion,
+    int eventOrdinal,
+    String partitionKey,
+    String contentType,
+    String correlationId,
+    String causationId,
+    Instant occurredAt,
+    String payload,
+    String payloadDigest) {
+
+  public ClaimedOutboxDelivery {
+    Objects.requireNonNull(owner, "owner");
+    Objects.requireNonNull(eventType, "eventType");
+    Objects.requireNonNull(aggregateType, "aggregateType");
+    Objects.requireNonNull(aggregateId, "aggregateId");
+    Objects.requireNonNull(partitionKey, "partitionKey");
+    Objects.requireNonNull(contentType, "contentType");
+    Objects.requireNonNull(correlationId, "correlationId");
+    Objects.requireNonNull(occurredAt, "occurredAt");
+    Objects.requireNonNull(payload, "payload");
+    Objects.requireNonNull(payloadDigest, "payloadDigest");
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/NewOutboxEventV2.java

@@ -0,0 +1,66 @@
+package dev.caskeleton.application.outbox.v2;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * Immutable outbox event intent pinned before any whole-transaction retry begins.
+ *
+ * <p>The aggregate version plus deterministic event ordinal is the ordering authority. The stable
+ * partition key is mandatory because this baseline advertises ordered destinations only.
+ */
+public record NewOutboxEventV2(
+    String eventId,
+    String aggregateType,
+    String aggregateId,
+    long aggregateVersion,
+    int eventOrdinal,
+    String eventType,
+    int eventSchema,
+    String logicalDestination,
+    String partitionKey,
+    String contentType,
+    String correlationId,
+    String causationId,
+    Instant occurredAt,
+    String payload) {
+
+  private static final int MAXIMUM_PAYLOAD_BYTES = 1024 * 1024;
+
+  public NewOutboxEventV2 {
+    requireBounded("event ID", eventId, 64);
+    requireBounded("aggregate type", aggregateType, 128);
+    requireBounded("aggregate ID", aggregateId, 256);
+    if (aggregateVersion < 1) {
+      throw new IllegalArgumentException("aggregate version must be positive");
+    }
+    if (eventOrdinal < 0 || eventOrdinal > 1023) {
+      throw new IllegalArgumentException("event ordinal must be between 0 and 1023");
+    }
+    requireBounded("event type", eventType, 256);
+    if (eventSchema < 1) {
+      throw new IllegalArgumentException("event schema must be positive");
+    }
+    requireBounded("logical destination", logicalDestination, 256);
+    requireBounded("partition key", partitionKey, 256);
+    requireBounded("content type", contentType, 128);
+    requireBounded("correlation ID", correlationId, 128);
+    if (causationId != null) {
+      requireBounded("causation ID", causationId, 128);
+    }
+    Objects.requireNonNull(occurredAt, "occurredAt");
+    Objects.requireNonNull(payload, "payload");
+    if (payload.isBlank()
+        || payload.getBytes(StandardCharsets.UTF_8).length > MAXIMUM_PAYLOAD_BYTES) {
+      throw new IllegalArgumentException(
+          "payload must be non-blank and at most " + MAXIMUM_PAYLOAD_BYTES + " UTF-8 bytes");
+    }
+  }
+
+  private static void requireBounded(String name, String value, int maximumLength) {
+    if (value == null || value.isBlank() || value.length() > maximumLength) {
+      throw new IllegalArgumentException(name + " must contain 1-" + maximumLength + " characters");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxAppendOutcome.java

@@ -0,0 +1,9 @@
+package dev.caskeleton.application.outbox.v2;
+
+/** Conflict-safe result of immutable identity plus event-envelope append. */
+public enum OutboxAppendOutcome {
+  APPENDED,
+  ALREADY_APPENDED_SAME_EVENT,
+  EVENT_ID_CONFLICT,
+  AGGREGATE_ORDER_CONFLICT
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxAppendPortV2.java

@@ -0,0 +1,12 @@
+package dev.caskeleton.application.outbox.v2;
+
+/**
+ * Same-store immutable outbox append.
+ *
+ * <p>The implementation must participate in the caller's active primary read-write transaction; it
+ * must never open a repository-local transaction.
+ */
+public interface OutboxAppendPortV2 {
+
+  OutboxAppendReceipt append(NewOutboxEventV2 event);
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxAppendReceipt.java

@@ -0,0 +1,23 @@
+package dev.caskeleton.application.outbox.v2;
+
+import java.time.LocalDate;
+import java.util.Objects;
+
+/** Database-authoritative routing receipt for one immutable event append. */
+public record OutboxAppendReceipt(
+    OutboxAppendOutcome outcome,
+    String eventId,
+    LocalDate retentionBucket,
+    long publicationEpoch,
+    OutboxDispatchAuthority dispatchAuthority) {
+
+  public OutboxAppendReceipt {
+    Objects.requireNonNull(outcome, "outcome");
+    Objects.requireNonNull(eventId, "eventId");
+    Objects.requireNonNull(retentionBucket, "retentionBucket");
+    Objects.requireNonNull(dispatchAuthority, "dispatchAuthority");
+    if (publicationEpoch < 1) {
+      throw new IllegalArgumentException("publication epoch must be positive");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxDeliveryClaimRequest.java

@@ -0,0 +1,26 @@
+package dev.caskeleton.application.outbox.v2;
+
+import java.time.Duration;
+
+/** Bounded strict-order polling claim request. */
+public record OutboxDeliveryClaimRequest(
+    String destination, String claimOwner, int batchSize, Duration claimLease) {
+
+  public OutboxDeliveryClaimRequest {
+    if (destination == null || destination.isBlank()) {
+      throw new IllegalArgumentException("destination must be present");
+    }
+    if (claimOwner == null || claimOwner.isBlank() || claimOwner.length() > 128) {
+      throw new IllegalArgumentException("claim owner must contain 1-128 characters");
+    }
+    if (batchSize < 1 || batchSize > 100) {
+      throw new IllegalArgumentException("batch size must be between 1 and 100");
+    }
+    if (claimLease == null
+        || claimLease.isZero()
+        || claimLease.isNegative()
+        || claimLease.compareTo(Duration.ofMinutes(5)) > 0) {
+      throw new IllegalArgumentException("claim lease must be positive and at most PT5M");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxDeliveryOwner.java

@@ -0,0 +1,45 @@
+package dev.caskeleton.application.outbox.v2;
+
+import java.time.LocalDate;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/** Full owner/token/status-version CAS handle for one destination delivery. */
+public record OutboxDeliveryOwner(
+    LocalDate retentionBucket,
+    String eventId,
+    String destination,
+    String claimOwner,
+    String claimToken,
+    int attempt,
+    long version,
+    long publicationEpoch) {
+
+  private static final Pattern CLAIM_TOKEN = Pattern.compile("[0-9a-f]{64}");
+
+  public OutboxDeliveryOwner {
+    Objects.requireNonNull(retentionBucket, "retentionBucket");
+    requirePresent(eventId, "event ID");
+    requirePresent(destination, "destination");
+    requirePresent(claimOwner, "claim owner");
+    if (claimToken == null || !CLAIM_TOKEN.matcher(claimToken).matches()) {
+      throw new IllegalArgumentException(
+          "claim token must be a 64-character lowercase hexadecimal value");
+    }
+    if (attempt < 1) {
+      throw new IllegalArgumentException("attempt must be positive");
+    }
+    if (version < 1) {
+      throw new IllegalArgumentException("version must be positive");
+    }
+    if (publicationEpoch < 1) {
+      throw new IllegalArgumentException("publication epoch must be positive");
+    }
+  }
+
+  private static void requirePresent(String value, String name) {
+    if (value == null || value.isBlank()) {
+      throw new IllegalArgumentException(name + " must be present");
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxDeliveryTransition.java

@@ -0,0 +1,13 @@
+package dev.caskeleton.application.outbox.v2;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.util.Objects;
+
+/** Idempotent transition request for a claimed delivery. */
+public record OutboxDeliveryTransition(OutboxDeliveryOwner owner, OperationId operationId) {
+
+  public OutboxDeliveryTransition {
+    Objects.requireNonNull(owner, "owner");
+    Objects.requireNonNull(operationId, "operationId");
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxDeliveryTransitionOutcome.java

@@ -0,0 +1,15 @@
+package dev.caskeleton.application.outbox.v2;
+
+/** Owner-safe polling completion/failure result. */
+public enum OutboxDeliveryTransitionOutcome {
+  PUBLISHED,
+  RETRY_SCHEDULED,
+  DEAD,
+  ALREADY_APPLIED_SAME_OPERATION,
+  RESULT_CONFLICT,
+  ABSENT,
+  NOT_OWNER,
+  NOT_CLAIMED,
+  STALE_VERSION,
+  AUTHORITY_MISMATCH
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxDispatchAuthority.java

@@ -0,0 +1,8 @@
+package dev.caskeleton.application.outbox.v2;
+
+/** Per-row dispatch authority derived from the locked publication control row. */
+public enum OutboxDispatchAuthority {
+  LEGACY_SHADOW,
+  POLLING_V2,
+  CDC
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxPollingDeliveryPortV2.java

@@ -0,0 +1,17 @@
+package dev.caskeleton.application.outbox.v2;
+
+import java.time.Instant;
+import java.util.List;
+
+/** Owner-safe polling delivery state machine; broker publish occurs outside its transactions. */
+public interface OutboxPollingDeliveryPortV2 {
+
+  List<ClaimedOutboxDelivery> claimBatch(OutboxDeliveryClaimRequest request);
+
+  OutboxDeliveryTransitionOutcome markPublished(OutboxDeliveryTransition transition);
+
+  OutboxDeliveryTransitionOutcome markRetryable(
+      OutboxDeliveryTransition transition, Instant nextAttemptAt, String errorCode);
+
+  OutboxDeliveryTransitionOutcome markDead(OutboxDeliveryTransition transition, String errorCode);
+}

src/application-core/src/main/java/dev/caskeleton/application/outbox/v2/OutboxPublicationAuthority.java

@@ -0,0 +1,8 @@
+package dev.caskeleton.application.outbox.v2;
+
+/** The single database-controlled publisher authority for the primary outbox scope. */
+public enum OutboxPublicationAuthority {
+  LEGACY_POLLING,
+  POLLING_V2,
+  CDC
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/OperationId.java

@@ -0,0 +1,26 @@
+package dev.caskeleton.application.transaction;
+
+import java.util.Objects;
+
+/**
+ * Stable, caller-owned identity for one logical write operation.
+ *
+ * <p>The value is intentionally opaque. Adapters may use it for reconciliation, but must not invent
+ * a replacement identity after an uncertain commit.
+ */
+public record OperationId(String value) {
+
+  private static final int MAXIMUM_LENGTH = 128;
+
+  public OperationId {
+    Objects.requireNonNull(value, "value must be non-null");
+    if (value.isBlank() || value.length() > MAXIMUM_LENGTH || !isPrintableAscii(value)) {
+      throw new IllegalArgumentException(
+          "operation ID must contain 1-128 printable non-whitespace ASCII characters");
+    }
+  }
+
+  private static boolean isPrintableAscii(String value) {
+    return value.chars().allMatch(character -> character >= 0x21 && character <= 0x7e);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/PolicyTransactionPort.java

@@ -0,0 +1,13 @@
+package dev.caskeleton.application.transaction;
+
+import java.util.function.Supplier;
+
+/**
+ * Additive transaction port for named policies and explicit outcomes.
+ *
+ * <p>{@link TransactionPort} remains source-compatible for existing callers and fakes.
+ */
+public interface PolicyTransactionPort extends TransactionPort {
+
+  <T> TransactionResult<T> inTransaction(TransactionRequest request, Supplier<T> action);
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/ReadConsistency.java

@@ -0,0 +1,9 @@
+package dev.caskeleton.application.transaction;
+
+/** Application-owned read consistency vocabulary. */
+public enum ReadConsistency {
+  STRONG,
+  READ_YOUR_WRITES,
+  EVENTUAL,
+  BOUNDED_STALENESS
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/ReconciliationReference.java

@@ -0,0 +1,24 @@
+package dev.caskeleton.application.transaction;
+
+import java.util.Objects;
+
+/**
+ * Sanitized, bounded reference that an operator or application workflow can use to reconcile an
+ * uncertain transaction outcome.
+ */
+public record ReconciliationReference(String value) {
+
+  private static final int MAXIMUM_LENGTH = 256;
+
+  public ReconciliationReference {
+    Objects.requireNonNull(value, "value must be non-null");
+    if (value.isBlank() || value.length() > MAXIMUM_LENGTH || !isPrintableAscii(value)) {
+      throw new IllegalArgumentException(
+          "reconciliation reference must contain 1-256 printable non-whitespace ASCII characters");
+    }
+  }
+
+  private static boolean isPrintableAscii(String value) {
+    return value.chars().allMatch(character -> character >= 0x21 && character <= 0x7e);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/TransactionAdmissionException.java

@@ -0,0 +1,16 @@
+package dev.caskeleton.application.transaction;
+
+/**
+ * A transaction was rejected before application work started because its policy, route, or
+ * remaining deadline could not be honored.
+ */
+public final class TransactionAdmissionException extends RuntimeException {
+
+  public TransactionAdmissionException(String message) {
+    super(message);
+  }
+
+  public TransactionAdmissionException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/TransactionOutcome.java

@@ -0,0 +1,10 @@
+package dev.caskeleton.application.transaction;
+
+/** Framework-neutral outcome of an application transaction boundary. */
+public enum TransactionOutcome {
+  COMMITTED,
+  PARTICIPATING_PENDING_OUTER,
+  DETERMINATE_ROLLBACK,
+  INDETERMINATE,
+  COMMITTED_WITH_POST_COMMIT_FAILURE
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/TransactionPhase.java

@@ -0,0 +1,12 @@
+package dev.caskeleton.application.transaction;
+
+/** Last safely observed phase of a physical transaction. */
+public enum TransactionPhase {
+  ROUTE_ADMISSION,
+  CONNECTION_ACQUIRED,
+  ACTIVE,
+  FLUSHED,
+  COMMIT_REQUESTED,
+  COMMIT_ACKED,
+  SYNCHRONIZATION_CLEANUP
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/TransactionPolicyId.java

@@ -0,0 +1,45 @@
+package dev.caskeleton.application.transaction;
+
+/**
+ * Allowlisted application transaction policies.
+ *
+ * <p>Callers select semantic policy IDs rather than framework propagation, isolation, route, or
+ * timeout numbers. Legacy facade policies remain adapter-internal and are intentionally absent.
+ */
+public enum TransactionPolicyId {
+  COMMAND_DEFAULT(false, true),
+  COMMAND_SERIALIZABLE_REPLAY_SAFE(false, true),
+  QUERY_PRIMARY(true, false),
+  QUERY_REPLICA_ELIGIBLE(true, false),
+  OUTBOX_APPEND(false, false),
+  INBOX_AND_HANDLER(false, true),
+  MAINTENANCE_NEW(false, true);
+
+  private final boolean readPolicy;
+  private final boolean operationIdRequired;
+
+  TransactionPolicyId(boolean readPolicy, boolean operationIdRequired) {
+    this.readPolicy = readPolicy;
+    this.operationIdRequired = operationIdRequired;
+  }
+
+  public boolean isReadPolicy() {
+    return readPolicy;
+  }
+
+  public boolean requiresOperationId() {
+    return operationIdRequired;
+  }
+
+  public boolean supports(ReadConsistency consistency) {
+    if (this == QUERY_PRIMARY) {
+      return consistency == ReadConsistency.STRONG
+          || consistency == ReadConsistency.READ_YOUR_WRITES;
+    }
+    if (this == QUERY_REPLICA_ELIGIBLE) {
+      return consistency == ReadConsistency.EVENTUAL
+          || consistency == ReadConsistency.BOUNDED_STALENESS;
+    }
+    return false;
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/TransactionRequest.java

@@ -0,0 +1,38 @@
+package dev.caskeleton.application.transaction;
+
+import dev.caskeleton.application.outbound.CallBudget;
+import java.util.Objects;
+import java.util.Optional;
+
+/** One framework-neutral request to execute an allowlisted transaction policy. */
+public record TransactionRequest(
+    TransactionPolicyId policyId,
+    CallBudget callBudget,
+    Optional<ReadConsistency> readConsistency,
+    Optional<OperationId> operationId) {
+
+  public TransactionRequest {
+    Objects.requireNonNull(policyId, "policyId must be non-null");
+    Objects.requireNonNull(callBudget, "callBudget must be non-null");
+    Objects.requireNonNull(readConsistency, "readConsistency must be non-null");
+    Objects.requireNonNull(operationId, "operationId must be non-null");
+
+    if (policyId.isReadPolicy()) {
+      ReadConsistency selected =
+          readConsistency.orElseThrow(
+              () ->
+                  new IllegalArgumentException(
+                      "readConsistency is required for " + policyId.name()));
+      if (!policyId.supports(selected)) {
+        throw new IllegalArgumentException(
+            "readConsistency " + selected + " is not allowed for " + policyId.name());
+      }
+    } else if (readConsistency.isPresent()) {
+      throw new IllegalArgumentException("readConsistency is forbidden for " + policyId.name());
+    }
+
+    if (policyId.requiresOperationId() && operationId.isEmpty()) {
+      throw new IllegalArgumentException("operationId is required for " + policyId.name());
+    }
+  }
+}

src/application-core/src/main/java/dev/caskeleton/application/transaction/TransactionResult.java

@@ -0,0 +1,83 @@
+package dev.caskeleton.application.transaction;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Outcome algebra for policy-based transactions.
+ *
+ * <p>A participant result never claims commit. An indeterminate result never grants replay
+ * authority.
+ */
+public sealed interface TransactionResult<T> {
+
+  TransactionOutcome outcome();
+
+  record Committed<T>(T value, Optional<OperationId> operationId) implements TransactionResult<T> {
+
+    public Committed {
+      Objects.requireNonNull(operationId, "operationId must be non-null");
+    }
+
+    @Override
+    public TransactionOutcome outcome() {
+      return TransactionOutcome.COMMITTED;
+    }
+  }
+
+  record Participating<T>(T value) implements TransactionResult<T> {
+
+    @Override
+    public TransactionOutcome outcome() {
+      return TransactionOutcome.PARTICIPATING_PENDING_OUTER;
+    }
+  }
+
+  record DeterminateRollback<T>(RuntimeException failure) implements TransactionResult<T> {
+
+    public DeterminateRollback {
+      Objects.requireNonNull(failure, "failure must be non-null");
+    }
+
+    @Override
+    public TransactionOutcome outcome() {
+      return TransactionOutcome.DETERMINATE_ROLLBACK;
+    }
+  }
+
+  record Indeterminate<T>(
+      Optional<OperationId> operationId,
+      TransactionPhase lastObservedPhase,
+      Optional<ReconciliationReference> reconciliationReference)
+      implements TransactionResult<T> {
+
+    public Indeterminate {
+      Objects.requireNonNull(operationId, "operationId must be non-null");
+      Objects.requireNonNull(lastObservedPhase, "lastObservedPhase must be non-null");
+      Objects.requireNonNull(reconciliationReference, "reconciliationReference must be non-null");
+      if (operationId.isEmpty() && reconciliationReference.isPresent()) {
+        throw new IllegalArgumentException("reconciliationReference requires a stable operationId");
+      }
+    }
+
+    @Override
+    public TransactionOutcome outcome() {
+      return TransactionOutcome.INDETERMINATE;
+    }
+  }
+
+  record CommittedWithPostCommitFailure<T>(
+      T value, Optional<OperationId> operationId, RuntimeException operationalFailure)
+      implements TransactionResult<T> {
+
+    public CommittedWithPostCommitFailure {
+      Objects.requireNonNull(operationId, "operationId must be non-null");
+      Objects.requireNonNull(operationalFailure, "operationalFailure must be non-null");
+    }
+
+    @Override
+    public TransactionOutcome outcome() {
+      return TransactionOutcome.COMMITTED_WITH_POST_COMMIT_FAILURE;
+    }
+  }
+}

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

@@ -0,0 +1,91 @@
+package dev.caskeleton.application.idempotency.v2;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.idempotency.RequestFingerprint;
+import dev.caskeleton.application.transaction.OperationId;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+
+class IdempotencyV2ContractTest {
+
+  private static final IdempotencyScopeDigest SCOPE =
+      new IdempotencyScopeDigest("a".repeat(64), 3, "CREATE_WORK_LOG");
+  private static final RequestFingerprint FINGERPRINT =
+      RequestFingerprint.ofSha256("request".getBytes(StandardCharsets.UTF_8));
+  private static final IdempotencyClaimAttempt ATTEMPT =
+      new IdempotencyClaimAttempt("b".repeat(64), new OperationId("claim-1"));
+
+  @Test
+  void scopeAcceptsOnlyCanonicalLowercaseSha256AndPositiveKeyVersion() {
+    assertThat(SCOPE.digest()).hasSize(64);
+
+    assertThatThrownBy(() -> new IdempotencyScopeDigest("A".repeat(64), 1, "CREATE_WORK_LOG"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("lowercase");
+    assertThatThrownBy(() -> new IdempotencyScopeDigest("a".repeat(64), 0, "CREATE_WORK_LOG"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("key digest version");
+  }
+
+  @Test
+  void claimRequestSeparatesFiniteProcessingAndReplayTtls() {
+    IdempotencyClaimRequest request =
+        new IdempotencyClaimRequest(
+            SCOPE,
+            FINGERPRINT,
+            ATTEMPT,
+            Duration.ofSeconds(30),
+            Duration.ofHours(24),
+            "json.v1",
+            2);
+
+    assertThat(request.processingLeaseTtl()).isEqualTo(Duration.ofSeconds(30));
+    assertThat(request.replayTtl()).isEqualTo(Duration.ofHours(24));
+
+    assertThatThrownBy(
+            () ->
+                new IdempotencyClaimRequest(
+                    SCOPE, FINGERPRINT, ATTEMPT, Duration.ZERO, Duration.ofHours(24), "json.v1", 2))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("processing lease TTL");
+  }
+
+  @Test
+  void ownerCarriesTheFullCasTuple() {
+    IdempotencyOwner owner =
+        new IdempotencyOwner(SCOPE, ATTEMPT.ownerToken(), 2, 7, ATTEMPT.operationId());
+
+    assertThat(owner.attempt()).isEqualTo(2);
+    assertThat(owner.stateRevision()).isEqualTo(7);
+    assertThat(owner.claimOperationId()).isEqualTo(new OperationId("claim-1"));
+
+    assertThatThrownBy(
+            () -> new IdempotencyOwner(SCOPE, ATTEMPT.ownerToken(), 0, 7, ATTEMPT.operationId()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("attempt");
+  }
+
+  @Test
+  void transitionResultRequiresAnOwnerOnlyForSuccessfulOwnerTransitions() {
+    IdempotencyOwner owner =
+        new IdempotencyOwner(SCOPE, ATTEMPT.ownerToken(), 1, 1, ATTEMPT.operationId());
+
+    assertThat(
+            new IdempotencyMutationResult<>(
+                    IdempotencyStartOutcome.STARTED, owner, IdempotencyStartOutcome::carriesOwner)
+                .owner())
+        .contains(owner);
+
+    assertThatThrownBy(
+            () ->
+                new IdempotencyMutationResult<>(
+                    IdempotencyStartOutcome.NOT_OWNER,
+                    owner,
+                    IdempotencyStartOutcome::carriesOwner))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("must not carry");
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/inbox/InboxContractTest.java

@@ -0,0 +1,39 @@
+package dev.caskeleton.application.inbox;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+
+class InboxContractTest {
+
+  @Test
+  void scopeIsAnOpaqueCanonicalDigestAndOwnerCarriesTheFullCasTuple() {
+    InboxScopeDigest scope = new InboxScopeDigest("a".repeat(64));
+    InboxOwner owner =
+        new InboxOwner(scope, "b".repeat(64), 2, 7, new OperationId("claim-message-1"));
+
+    assertThat(owner.attempt()).isEqualTo(2);
+    assertThat(owner.stateRevision()).isEqualTo(7);
+
+    assertThatThrownBy(() -> new InboxScopeDigest("A".repeat(64)))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("lowercase");
+  }
+
+  @Test
+  void claimRequestSeparatesProcessingLeaseFromTerminalRetention() {
+    InboxClaimRequest request =
+        new InboxClaimRequest(
+            new InboxScopeDigest("a".repeat(64)),
+            "c".repeat(64),
+            new InboxClaimAttempt("b".repeat(64), new OperationId("claim-message-1")),
+            Duration.ofSeconds(30),
+            Duration.ofDays(7));
+
+    assertThat(request.processingLease()).isEqualTo(Duration.ofSeconds(30));
+    assertThat(request.terminalRetention()).isEqualTo(Duration.ofDays(7));
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/outbound/CallBudgetTest.java

@@ -33,4 +33,11 @@ class CallBudgetTest {
     assertThatThrownBy(() -> CallBudget.after(1_000, Duration.ofDays(366)))
         .isInstanceOf(IllegalArgumentException.class);
   }
+
+  @Test
+  void rejectsMonotonicDeadlineOverflow() {
+    assertThatThrownBy(() -> CallBudget.after(Long.MAX_VALUE - 10, Duration.ofNanos(11)))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("deadline");
+  }
 }

src/application-core/src/test/java/dev/caskeleton/application/outbox/v2/OutboxDeliveryV2ContractTest.java

@@ -0,0 +1,61 @@
+package dev.caskeleton.application.outbox.v2;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.transaction.OperationId;
+import java.time.LocalDate;
+import org.junit.jupiter.api.Test;
+
+class OutboxDeliveryV2ContractTest {
+
+  @Test
+  void ownerHandleCarriesTheFullDeliveryCasTuple() {
+    OutboxDeliveryOwner owner =
+        new OutboxDeliveryOwner(
+            LocalDate.parse("2026-07-28"),
+            "event-1",
+            "portfolio.events",
+            "relay-1",
+            "a".repeat(64),
+            2,
+            7,
+            3);
+
+    assertThat(owner.attempt()).isEqualTo(2);
+    assertThat(owner.version()).isEqualTo(7);
+    assertThat(owner.publicationEpoch()).isEqualTo(3);
+
+    assertThatThrownBy(
+            () ->
+                new OutboxDeliveryOwner(
+                    LocalDate.parse("2026-07-28"),
+                    "event-1",
+                    "portfolio.events",
+                    "relay-1",
+                    "a".repeat(64),
+                    0,
+                    7,
+                    3))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("attempt");
+  }
+
+  @Test
+  void completionRequestRequiresStableOperationIdentity() {
+    OutboxDeliveryTransition transition =
+        new OutboxDeliveryTransition(
+            new OutboxDeliveryOwner(
+                LocalDate.parse("2026-07-28"),
+                "event-1",
+                "portfolio.events",
+                "relay-1",
+                "a".repeat(64),
+                1,
+                2,
+                1),
+            new OperationId("publish-complete-1"));
+
+    assertThat(transition.operationId().value()).isEqualTo("publish-complete-1");
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/outbox/v2/OutboxV2ContractTest.java

@@ -0,0 +1,76 @@
+package dev.caskeleton.application.outbox.v2;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.time.Instant;
+import org.junit.jupiter.api.Test;
+
+class OutboxV2ContractTest {
+
+  @Test
+  void eventRequiresPinnedAggregateVersionOrdinalAndBoundedEnvelopeIdentity() {
+    NewOutboxEventV2 event =
+        new NewOutboxEventV2(
+            "event-1",
+            "WorkLog",
+            "work-log-42",
+            7,
+            0,
+            "WorkLogCreated",
+            1,
+            "portfolio.events",
+            "work-log-42",
+            "application/json",
+            "correlation-1",
+            null,
+            Instant.parse("2026-07-28T12:00:00Z"),
+            "{\"id\":\"42\"}");
+
+    assertThat(event.aggregateVersion()).isEqualTo(7);
+    assertThat(event.eventOrdinal()).isZero();
+
+    assertThatThrownBy(
+            () ->
+                new NewOutboxEventV2(
+                    "event-1",
+                    "WorkLog",
+                    "work-log-42",
+                    0,
+                    0,
+                    "WorkLogCreated",
+                    1,
+                    "portfolio.events",
+                    "work-log-42",
+                    "application/json",
+                    "correlation-1",
+                    null,
+                    Instant.parse("2026-07-28T12:00:00Z"),
+                    "{}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("aggregate version");
+  }
+
+  @Test
+  void orderedDestinationDescriptorRequiresAStablePartitionKey() {
+    assertThatThrownBy(
+            () ->
+                new NewOutboxEventV2(
+                    "event-1",
+                    "WorkLog",
+                    "work-log-42",
+                    1,
+                    0,
+                    "WorkLogCreated",
+                    1,
+                    "portfolio.events",
+                    null,
+                    "application/json",
+                    "correlation-1",
+                    null,
+                    Instant.parse("2026-07-28T12:00:00Z"),
+                    "{}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("partition key");
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/transaction/OperationIdTest.java

@@ -0,0 +1,25 @@
+package dev.caskeleton.application.transaction;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+class OperationIdTest {
+
+  @Test
+  void preservesAnOpaqueStableIdentifier() {
+    OperationId operationId = new OperationId("job-20260728-item-42");
+
+    assertThat(operationId.value()).isEqualTo("job-20260728-item-42");
+  }
+
+  @Test
+  void rejectsBlankControlCharactersAndUnboundedValues() {
+    assertThatThrownBy(() -> new OperationId(" ")).isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new OperationId("operation\nid"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new OperationId("a".repeat(129)))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/transaction/TransactionRequestTest.java

@@ -0,0 +1,114 @@
+package dev.caskeleton.application.transaction;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import dev.caskeleton.application.outbound.CallBudget;
+import java.time.Duration;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+
+class TransactionRequestTest {
+
+  private static final CallBudget BUDGET = CallBudget.after(1_000, Duration.ofSeconds(5));
+  private static final OperationId OPERATION_ID = new OperationId("operation-42");
+
+  @Test
+  void commandPoliciesRequireAStableOperationIdAndRejectReadConsistency() {
+    TransactionRequest request =
+        new TransactionRequest(
+            TransactionPolicyId.COMMAND_DEFAULT,
+            BUDGET,
+            Optional.empty(),
+            Optional.of(OPERATION_ID));
+
+    assertThat(request.operationId()).contains(OPERATION_ID);
+    assertThatThrownBy(
+            () ->
+                new TransactionRequest(
+                    TransactionPolicyId.COMMAND_DEFAULT,
+                    BUDGET,
+                    Optional.empty(),
+                    Optional.empty()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("operationId");
+    assertThatThrownBy(
+            () ->
+                new TransactionRequest(
+                    TransactionPolicyId.COMMAND_DEFAULT,
+                    BUDGET,
+                    Optional.of(ReadConsistency.STRONG),
+                    Optional.of(OPERATION_ID)))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("readConsistency");
+  }
+
+  @Test
+  void queryPrimaryAcceptsOnlyStrongOrReadYourWrites() {
+    assertThat(
+            new TransactionRequest(
+                    TransactionPolicyId.QUERY_PRIMARY,
+                    BUDGET,
+                    Optional.of(ReadConsistency.STRONG),
+                    Optional.empty())
+                .readConsistency())
+        .contains(ReadConsistency.STRONG);
+    assertThat(
+            new TransactionRequest(
+                    TransactionPolicyId.QUERY_PRIMARY,
+                    BUDGET,
+                    Optional.of(ReadConsistency.READ_YOUR_WRITES),
+                    Optional.empty())
+                .readConsistency())
+        .contains(ReadConsistency.READ_YOUR_WRITES);
+    assertThatThrownBy(
+            () ->
+                new TransactionRequest(
+                    TransactionPolicyId.QUERY_PRIMARY,
+                    BUDGET,
+                    Optional.of(ReadConsistency.EVENTUAL),
+                    Optional.empty()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("QUERY_PRIMARY");
+  }
+
+  @Test
+  void replicaEligibleQueryAcceptsOnlyEventualOrBoundedStaleness() {
+    assertThat(
+            new TransactionRequest(
+                    TransactionPolicyId.QUERY_REPLICA_ELIGIBLE,
+                    BUDGET,
+                    Optional.of(ReadConsistency.BOUNDED_STALENESS),
+                    Optional.empty())
+                .readConsistency())
+        .contains(ReadConsistency.BOUNDED_STALENESS);
+    assertThatThrownBy(
+            () ->
+                new TransactionRequest(
+                    TransactionPolicyId.QUERY_REPLICA_ELIGIBLE,
+                    BUDGET,
+                    Optional.of(ReadConsistency.STRONG),
+                    Optional.empty()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("QUERY_REPLICA_ELIGIBLE");
+  }
+
+  @Test
+  void readPoliciesRequireAnExplicitConsistency() {
+    assertThatThrownBy(
+            () ->
+                new TransactionRequest(
+                    TransactionPolicyId.QUERY_PRIMARY, BUDGET, Optional.empty(), Optional.empty()))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("readConsistency");
+  }
+
+  @Test
+  void outboxAppendCanInheritTheOuterOperationIdentity() {
+    TransactionRequest request =
+        new TransactionRequest(
+            TransactionPolicyId.OUTBOX_APPEND, BUDGET, Optional.empty(), Optional.empty());
+
+    assertThat(request.operationId()).isEmpty();
+  }
+}

src/application-core/src/test/java/dev/caskeleton/application/transaction/TransactionResultTest.java

@@ -0,0 +1,58 @@
+package dev.caskeleton.application.transaction;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+
+class TransactionResultTest {
+
+  private static final OperationId OPERATION_ID = new OperationId("operation-42");
+
+  @Test
+  void representsCommittedAndParticipatingResultsWithoutConflatingThem() {
+    TransactionResult<String> committed =
+        new TransactionResult.Committed<>("value", Optional.of(OPERATION_ID));
+    TransactionResult<String> participating = new TransactionResult.Participating<>("value");
+
+    assertThat(committed.outcome()).isEqualTo(TransactionOutcome.COMMITTED);
+    assertThat(participating.outcome()).isEqualTo(TransactionOutcome.PARTICIPATING_PENDING_OUTER);
+  }
+
+  @Test
+  void indeterminateOutcomeCarriesOnlyOptionalSafeReconciliationData() {
+    ReconciliationReference reference =
+        new ReconciliationReference("operation-ledger:operation-42");
+    TransactionResult<String> result =
+        new TransactionResult.Indeterminate<>(
+            Optional.of(OPERATION_ID), TransactionPhase.COMMIT_REQUESTED, Optional.of(reference));
+
+    assertThat(result.outcome()).isEqualTo(TransactionOutcome.INDETERMINATE);
+    assertThat(((TransactionResult.Indeterminate<String>) result).reconciliationReference())
+        .contains(reference);
+  }
+
+  @Test
+  void determinateRollbackAndPostCommitFailureKeepDifferentOutcomes() {
+    RuntimeException rollback = new IllegalStateException("rolled back");
+    RuntimeException cleanup = new IllegalStateException("cleanup failed");
+
+    TransactionResult<String> rolledBack = new TransactionResult.DeterminateRollback<>(rollback);
+    TransactionResult<String> committedWithFailure =
+        new TransactionResult.CommittedWithPostCommitFailure<>(
+            "value", Optional.of(OPERATION_ID), cleanup);
+
+    assertThat(rolledBack.outcome()).isEqualTo(TransactionOutcome.DETERMINATE_ROLLBACK);
+    assertThat(committedWithFailure.outcome())
+        .isEqualTo(TransactionOutcome.COMMITTED_WITH_POST_COMMIT_FAILURE);
+  }
+
+  @Test
+  void reconciliationReferenceRejectsUnsafeOrUnboundedText() {
+    assertThatThrownBy(() -> new ReconciliationReference("raw\nsql"))
+        .isInstanceOf(IllegalArgumentException.class);
+    assertThatThrownBy(() -> new ReconciliationReference("a".repeat(257)))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+}