feat: web, websocket 어댑터 추가 구현

2026-08-28 17:01:27 +09:00
parent 0137263441
commit a24ece9cf7
883 changed files with 100584 additions and 2623 deletions
@@ -1,7 +1,8 @@
 # This is a Gradle generated file for dependency locking.
 # Manual edits can break the build and are not advised.
 # This file is expected to be part of source control.
-com.github.ben-manes.caffeine:caffeine:3.2.3=annotationProcessor,redisPolicyContractTestAnnotationProcessor,testAnnotationProcessor
+com.github.ben-manes.caffeine:caffeine:3.2.3=redisPolicyContractTestAnnotationProcessor
+com.github.ben-manes.caffeine:caffeine:3.2.4=annotationProcessor,testAnnotationProcessor
 com.github.kevinstern:software-and-algorithms:1.0=annotationProcessor,redisPolicyContractTestAnnotationProcessor,testAnnotationProcessor
 com.github.spotbugs:spotbugs-annotations:4.10.2=spotbugs
 com.github.spotbugs:spotbugs:4.10.2=spotbugs
@@ -33,7 +34,7 @@ info.picocli:picocli:4.7.7=checkstyle
 io.github.eisop:dataflow-errorprone:3.41.0-eisop1=annotationProcessor,redisPolicyContractTestAnnotationProcessor,testAnnotationProcessor
 io.github.java-diff-utils:java-diff-utils:4.12=annotationProcessor,redisPolicyContractTestAnnotationProcessor,testAnnotationProcessor
 javax.inject:javax.inject:1=annotationProcessor,redisPolicyContractTestAnnotationProcessor,testAnnotationProcessor
-jaxen:jaxen:2.0.0=spotbugs
+jaxen:jaxen:2.0.6=spotbugs
 net.bytebuddy:byte-buddy:1.17.8=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
 net.jqwik:jqwik-api:1.9.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
 net.jqwik:jqwik-engine:1.9.1=redisPolicyContractTestRuntimeClasspath,testRuntimeClasspath
@@ -48,30 +49,40 @@ org.apache.commons:commons-text:1.15.0=spotbugs
 org.apache.commons:commons-text:1.3=checkstyle
 org.apache.httpcomponents:httpclient:4.5.13=checkstyle
 org.apache.httpcomponents:httpcore:4.4.16=checkstyle
-org.apache.logging.log4j:log4j-api:2.25.2=spotbugs
-org.apache.logging.log4j:log4j-core:2.25.2=spotbugs
+org.apache.logging.log4j:log4j-api:2.25.5=spotbugs
+org.apache.logging.log4j:log4j-core:2.25.5=spotbugs
 org.apache.maven.doxia:doxia-core:1.12.0=checkstyle
 org.apache.maven.doxia:doxia-logging-api:1.12.0=checkstyle
 org.apache.maven.doxia:doxia-module-xdoc:1.12.0=checkstyle
 org.apache.maven.doxia:doxia-sink-api:1.12.0=checkstyle
 org.apache.xbean:xbean-reflect:3.7=checkstyle
 org.apiguardian:apiguardian-api:1.1.2=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
-org.assertj:assertj-core:3.27.6=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.assertj:assertj-core:3.27.6=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath
+org.assertj:assertj-core:3.27.7=testCompileClasspath,testRuntimeClasspath
 org.codehaus.plexus:plexus-classworlds:2.6.0=checkstyle
 org.codehaus.plexus:plexus-component-annotations:2.1.0=checkstyle
 org.codehaus.plexus:plexus-container-default:2.1.0=checkstyle
 org.codehaus.plexus:plexus-utils:3.3.0=checkstyle
 org.dom4j:dom4j:2.2.0=spotbugs
 org.javassist:javassist:3.28.0-GA=checkstyle
-org.jspecify:jspecify:1.0.0=annotationProcessor,checkstyle,redisPolicyContractTestAnnotationProcessor,redisPolicyContractTestCompileClasspath,testAnnotationProcessor,testCompileClasspath
-org.junit.jupiter:junit-jupiter-api:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
-org.junit.jupiter:junit-jupiter-engine:6.0.1=redisPolicyContractTestRuntimeClasspath,testRuntimeClasspath
-org.junit.jupiter:junit-jupiter-params:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
-org.junit.jupiter:junit-jupiter:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
-org.junit.platform:junit-platform-commons:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
-org.junit.platform:junit-platform-engine:6.0.1=redisPolicyContractTestRuntimeClasspath,testRuntimeClasspath
-org.junit.platform:junit-platform-launcher:6.0.1=redisPolicyContractTestRuntimeClasspath,testRuntimeClasspath
-org.junit:junit-bom:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
+org.jspecify:jspecify:1.0.0=redisPolicyContractTestAnnotationProcessor,redisPolicyContractTestCompileClasspath
+org.jspecify:jspecify:1.0.1=annotationProcessor,checkstyle,testAnnotationProcessor,testCompileClasspath
+org.junit.jupiter:junit-jupiter-api:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath
+org.junit.jupiter:junit-jupiter-api:6.0.3=testCompileClasspath,testRuntimeClasspath
+org.junit.jupiter:junit-jupiter-engine:6.0.1=redisPolicyContractTestRuntimeClasspath
+org.junit.jupiter:junit-jupiter-engine:6.0.3=testRuntimeClasspath
+org.junit.jupiter:junit-jupiter-params:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath
+org.junit.jupiter:junit-jupiter-params:6.0.3=testCompileClasspath,testRuntimeClasspath
+org.junit.jupiter:junit-jupiter:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath
+org.junit.jupiter:junit-jupiter:6.0.3=testCompileClasspath,testRuntimeClasspath
+org.junit.platform:junit-platform-commons:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath
+org.junit.platform:junit-platform-commons:6.0.3=testCompileClasspath,testRuntimeClasspath
+org.junit.platform:junit-platform-engine:6.0.1=redisPolicyContractTestRuntimeClasspath
+org.junit.platform:junit-platform-engine:6.0.3=testRuntimeClasspath
+org.junit.platform:junit-platform-launcher:6.0.1=redisPolicyContractTestRuntimeClasspath
+org.junit.platform:junit-platform-launcher:6.0.3=testRuntimeClasspath
+org.junit:junit-bom:6.0.1=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath
+org.junit:junit-bom:6.0.3=testCompileClasspath,testRuntimeClasspath
 org.junit:junit-bom:6.1.0=spotbugs
 org.mockito:mockito-core:5.20.0=mockitoAgent
 org.opentest4j:opentest4j:1.3.0=redisPolicyContractTestCompileClasspath,redisPolicyContractTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath
@@ -82,7 +93,8 @@ org.ow2.asm:asm-util:9.10.1=spotbugs
 org.ow2.asm:asm:9.10.1=spotbugs
 org.pcollections:pcollections:4.0.1=annotationProcessor,redisPolicyContractTestAnnotationProcessor,testAnnotationProcessor
 org.reflections:reflections:0.10.2=checkstyle
-org.slf4j:slf4j-api:2.0.17=spotbugs,spotbugsSlf4j
-org.slf4j:slf4j-simple:2.0.17=checkstyle,spotbugsSlf4j
+org.slf4j:slf4j-api:2.0.18=spotbugs,spotbugsSlf4j
+org.slf4j:slf4j-simple:2.0.17=spotbugsSlf4j
+org.slf4j:slf4j-simple:2.0.18=checkstyle
 org.xmlresolver:xmlresolver:5.3.3=checkstyle,spotbugs
 empty=compileClasspath,runtimeClasspath
@@ -0,0 +1,100 @@
+package dev.caskeleton.application.operation;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * One durable operation as stored.
+ *
+ * <p>Durable is the point. A {@code CompletableFuture} or a scheduler handle is process-local: it
+ * disappears with the node that made it, and a caller polling for it is told the work does not
+ * exist while it is still running elsewhere. Everything in this record survives a restart.
+ *
+ * <p>The constructor refuses states that cannot be true at once. Each is a specific way a poller
+ * ends up unable to act: a SUCCEEDED operation with no result reference, a FAILED one with no
+ * recorded reason, a RUNNING one with no lease.
+ *
+ * @param operationId the identity callers poll
+ * @param operationName what work this is
+ * @param principal who submitted it
+ * @param tenantId the tenant, or null when the operation is not tenant scoped
+ * @param state where in the lifecycle it is
+ * @param submittedAt when it was accepted
+ * @param startedAt when a worker first picked it up
+ * @param completedAt when it reached a terminal state
+ * @param progress how far it has got
+ * @param resultReference where a successful result can be read
+ * @param failure why it failed
+ * @param lease the worker's claim, while one is held
+ * @param expiresAt when this record stops being available
+ */
+public record DurableOperation(
+    DurableOperationId operationId,
+    String operationName,
+    String principal,
+    String tenantId,
+    DurableOperationState state,
+    Instant submittedAt,
+    Optional<Instant> startedAt,
+    Optional<Instant> completedAt,
+    Optional<OperationProgressSnapshot> progress,
+    Optional<String> resultReference,
+    Optional<OperationFailure> failure,
+    Optional<OperationLease> lease,
+    Instant expiresAt) {
+
+  public DurableOperation {
+    Objects.requireNonNull(operationId, "operationId");
+    Objects.requireNonNull(operationName, "operationName");
+    Objects.requireNonNull(principal, "principal");
+    Objects.requireNonNull(state, "state");
+    Objects.requireNonNull(submittedAt, "submittedAt");
+    Objects.requireNonNull(startedAt, "startedAt");
+    Objects.requireNonNull(completedAt, "completedAt");
+    Objects.requireNonNull(progress, "progress");
+    Objects.requireNonNull(resultReference, "resultReference");
+    Objects.requireNonNull(failure, "failure");
+    Objects.requireNonNull(lease, "lease");
+    Objects.requireNonNull(expiresAt, "expiresAt");
+
+    if (state == DurableOperationState.SUCCEEDED && resultReference.isEmpty()) {
+      throw new IllegalArgumentException(
+          "a SUCCEEDED operation must record where its result is; a success nobody can read is"
+              + " indistinguishable from a lost one");
+    }
+    if (state == DurableOperationState.FAILED && failure.isEmpty()) {
+      throw new IllegalArgumentException(
+          "a FAILED operation must record why; a failure with no reason gives the caller nothing"
+              + " to report and nothing to decide on");
+    }
+    if (state != DurableOperationState.FAILED && failure.isPresent()) {
+      throw new IllegalArgumentException("only a FAILED operation records a failure");
+    }
+    if (state.terminal() && completedAt.isEmpty()) {
+      throw new IllegalArgumentException("a " + state + " operation must record when it finished");
+    }
+    if (!state.terminal() && completedAt.isPresent()) {
+      throw new IllegalArgumentException("a " + state + " operation has not finished");
+    }
+    if (state == DurableOperationState.RUNNING && lease.isEmpty()) {
+      throw new IllegalArgumentException(
+          "a RUNNING operation must hold a lease; without one nothing can tell a live worker from"
+              + " a dead one, and the operation is stranded");
+    }
+    if (state.terminal() && lease.isPresent()) {
+      throw new IllegalArgumentException(
+          "a finished operation still holding a lease blocks reclaim");
+    }
+  }
+
+  /** Whether a poller should stop polling. */
+  public boolean finished() {
+    return state.terminal();
+  }
+
+  /** Whether the lease, if any, has lapsed and the operation can be reclaimed. */
+  public boolean leaseExpiredAt(Instant now) {
+    return lease.map(held -> held.expiredAt(now)).orElse(false);
+  }
+}
@@ -0,0 +1,30 @@
+package dev.caskeleton.application.operation;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * The identity of one durable operation.
+ *
+ * @param value the identifier
+ */
+public record DurableOperationId(String value) {
+
+  // Bounded and free of anything that needs escaping. The identifier travels to whatever transport
+  // published the operation and is handed back to callers, so a value that one encoding mangles is
+  // a defect that only appears for some clients.
+  private static final Pattern GRAMMAR = Pattern.compile("[A-Za-z0-9._~-]{1,128}");
+
+  public DurableOperationId {
+    Objects.requireNonNull(value, "value");
+    if (!GRAMMAR.matcher(value).matches()) {
+      throw new IllegalArgumentException(
+          "a durable operation id must match [A-Za-z0-9._~-]{1,128}");
+    }
+  }
+
+  @Override
+  public String toString() {
+    return value;
+  }
+}
@@ -0,0 +1,38 @@
+package dev.caskeleton.application.operation;
+
+/**
+ * The lifecycle of a durable operation as the application sees it.
+ *
+ * <p>Deliberately separate from any transport's rendering of it. A transport may choose to expose
+ * fewer states or different names; it may not invent a state the store cannot be in.
+ */
+public enum DurableOperationState {
+
+  /** Accepted and durable, waiting for a worker. */
+  PENDING,
+
+  /** A worker holds an unexpired lease. */
+  RUNNING,
+
+  /** Finished; a result reference is recorded. */
+  SUCCEEDED,
+
+  /** Finished; a failure is recorded and no retry will happen. */
+  FAILED,
+
+  /** Stopped on request before it finished. */
+  CANCELED,
+
+  /** The record aged out. The work may have succeeded; the record no longer says. */
+  EXPIRED;
+
+  /** Whether no further transition is possible. */
+  public boolean terminal() {
+    return this == SUCCEEDED || this == FAILED || this == CANCELED || this == EXPIRED;
+  }
+
+  /** Whether cancellation is still meaningful. */
+  public boolean cancellable() {
+    return this == PENDING || this == RUNNING;
+  }
+}
@@ -0,0 +1,111 @@
+package dev.caskeleton.application.operation;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Where durable operations live.
+ *
+ * <p>A port because the application owns what an operation is and nothing about how it is stored.
+ * The JPA adapter behind it is one implementation; an in-memory one serves tests.
+ *
+ * <p>Every method that changes a running operation takes the {@code workerId} and verifies it. A
+ * worker whose lease lapsed may still be alive and still finishing its work, and letting it report
+ * a result after another worker took over is how an operation ends up with two answers.
+ */
+public interface DurableOperationStorePort {
+
+  /**
+   * Records an accepted submission, or returns the operation an identical one already created.
+   *
+   * <p>Identical means the same scope and the same fingerprint. Returning the existing operation
+   * rather than a second one is what makes a resubmission safe while the first is still running.
+   *
+   * @param submission what was accepted
+   * @return the operation, whether newly created or already present
+   */
+  DurableOperation submit(DurableOperationSubmission submission);
+
+  /**
+   * Reads an operation the caller is entitled to see.
+   *
+   * @param operationId the identity being polled
+   * @param principal the caller
+   * @param now the instant expiry is judged against
+   */
+  Optional<DurableOperation> find(DurableOperationId operationId, String principal, Instant now);
+
+  /**
+   * Takes a lease on the next runnable operation.
+   *
+   * <p>Runnable means PENDING, or RUNNING with a lapsed lease. The second case is the recovery
+   * path: without it, an operation claimed by a worker that then died is never picked up again.
+   *
+   * @param workerId who is claiming
+   * @param leaseDuration how long the claim holds without a heartbeat
+   * @param now the current instant
+   */
+  Optional<DurableOperationSubmission> claimNext(
+      String workerId, Duration leaseDuration, Instant now);
+
+  /**
+   * Extends a lease the worker still holds.
+   *
+   * @return false when the lease is no longer this worker's, which means it must stop
+   */
+  boolean heartbeat(
+      DurableOperationId operationId, String workerId, Duration leaseDuration, Instant now);
+
+  /**
+   * Publishes progress for an operation this worker holds.
+   *
+   * @return false when the lease is no longer this worker's
+   */
+  boolean reportProgress(
+      DurableOperationId operationId, String workerId, OperationProgressSnapshot progress);
+
+  /**
+   * Records success.
+   *
+   * @return false when the lease is no longer this worker's, so the result is not recorded
+   */
+  boolean succeed(
+      DurableOperationId operationId, String workerId, String resultReference, Instant completedAt);
+
+  /**
+   * Records failure.
+   *
+   * @return false when the lease is no longer this worker's
+   */
+  boolean fail(
+      DurableOperationId operationId,
+      String workerId,
+      OperationFailure failure,
+      Instant completedAt);
+
+  /**
+   * Requests cancellation.
+   *
+   * <p>Idempotent, and never reverses a terminal state: cancelling an operation that already
+   * succeeded would replace a real answer with a lie about it.
+   *
+   * @return whether this call moved the operation to CANCELED
+   */
+  boolean cancel(DurableOperationId operationId, String principal, Instant canceledAt);
+
+  /**
+   * Returns operations whose lease lapsed to PENDING.
+   *
+   * @return the operations reclaimed
+   */
+  List<DurableOperationId> reclaimExpiredLeases(Instant now);
+
+  /**
+   * Marks records that outlived their TTL as EXPIRED.
+   *
+   * @return the operations expired
+   */
+  List<DurableOperationId> expireStaleOperations(Instant now);
+}
@@ -0,0 +1,62 @@
+package dev.caskeleton.application.operation;
+
+import dev.caskeleton.application.idempotency.RequestFingerprint;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * What is durably written when long-running work is accepted.
+ *
+ * <p>Separate from {@link DurableOperation} because they have different readers: this is written
+ * once at admission and carries the payload a worker will execute; the operation is read on every
+ * poll. Folding them together would put the submitted payload into the document handed back to
+ * every poller.
+ *
+ * <p>It carries the request fingerprint so a resubmission under the same key returns the operation
+ * already accepted rather than starting a second one. An accepted-but-unfinished operation is
+ * precisely the state in which a client is most likely to retry.
+ *
+ * @param operationId the identity handed back
+ * @param operationName what work this is
+ * @param principal who submitted it
+ * @param tenantId the tenant, or null when the operation is not tenant scoped
+ * @param fingerprint the semantic fingerprint of the submitted request
+ * @param payload the serialized request a worker will execute
+ * @param submittedAt when it was accepted
+ * @param timeToLive how long the record and its result stay available
+ */
+public record DurableOperationSubmission(
+    DurableOperationId operationId,
+    String operationName,
+    String principal,
+    String tenantId,
+    RequestFingerprint fingerprint,
+    String payload,
+    Instant submittedAt,
+    Duration timeToLive) {
+
+  public DurableOperationSubmission {
+    Objects.requireNonNull(operationId, "operationId");
+    Objects.requireNonNull(operationName, "operationName");
+    Objects.requireNonNull(fingerprint, "fingerprint");
+    Objects.requireNonNull(payload, "payload");
+    Objects.requireNonNull(submittedAt, "submittedAt");
+    Objects.requireNonNull(timeToLive, "timeToLive");
+    if (principal == null || principal.isBlank()) {
+      throw new IllegalArgumentException(
+          "a submission needs a principal; without one any caller could poll any operation");
+    }
+    if (operationName.isBlank()) {
+      throw new IllegalArgumentException("a submission needs the operation it is for");
+    }
+    if (timeToLive.isZero() || timeToLive.isNegative()) {
+      throw new IllegalArgumentException("a submission with no lifetime cannot be polled");
+    }
+  }
+
+  /** When this operation's record stops being available. */
+  public Instant expiresAt() {
+    return submittedAt.plus(timeToLive);
+  }
+}
@@ -0,0 +1,26 @@
+package dev.caskeleton.application.operation;
+
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * What the caller is told after a submission commits.
+ *
+ * <p>{@code alreadyAccepted} is the part that matters. A resubmission of identical work returns the
+ * operation that already exists, and the transport needs to know which case it is: a first
+ * acceptance is a 202 with a fresh {@code Location}, and a repeat is a 202 pointing at the same
+ * operation. Without the flag the transport cannot tell, and the honest thing — pointing the client
+ * at the work that is actually running — becomes indistinguishable from silently starting a second.
+ *
+ * @param operationId the identity the caller polls
+ * @param acceptedAt when the operation was first accepted, not when this call happened
+ * @param alreadyAccepted whether an identical submission had already created it
+ */
+public record OperationAcceptance(
+    DurableOperationId operationId, Instant acceptedAt, boolean alreadyAccepted) {
+
+  public OperationAcceptance {
+    Objects.requireNonNull(operationId, "operationId");
+    Objects.requireNonNull(acceptedAt, "acceptedAt");
+  }
+}
@@ -0,0 +1,52 @@
+package dev.caskeleton.application.operation;
+
+import dev.caskeleton.application.idempotency.RequestFingerprint;
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * A request to run work asynchronously.
+ *
+ * <p>The caller supplies the operation id rather than the store minting one. That is what lets the
+ * transport hand back a {@code Location} it already knows, and it makes the whole submission one
+ * value the caller can retry: a generated id would differ on every attempt, so a retry after an
+ * ambiguous failure would create a second operation.
+ *
+ * @param operationId the identity to use
+ * @param operationName what work this is
+ * @param principal who is asking
+ * @param tenantId the tenant, or null when the operation is not tenant scoped
+ * @param fingerprint the semantic fingerprint of the request
+ * @param payload the serialized request a worker will execute
+ * @param eventType the outbox event type a worker subscribes to
+ * @param timeToLive how long the record and its result stay available
+ */
+public record OperationCommand(
+    DurableOperationId operationId,
+    String operationName,
+    String principal,
+    String tenantId,
+    RequestFingerprint fingerprint,
+    String payload,
+    String eventType,
+    Duration timeToLive) {
+
+  public OperationCommand {
+    Objects.requireNonNull(operationId, "operationId");
+    Objects.requireNonNull(operationName, "operationName");
+    Objects.requireNonNull(fingerprint, "fingerprint");
+    Objects.requireNonNull(payload, "payload");
+    Objects.requireNonNull(timeToLive, "timeToLive");
+    if (principal == null || principal.isBlank()) {
+      throw new IllegalArgumentException("an operation command needs a principal");
+    }
+    if (operationName.isBlank()) {
+      throw new IllegalArgumentException("an operation command needs the operation it is for");
+    }
+    if (eventType == null || eventType.isBlank()) {
+      throw new IllegalArgumentException(
+          "an operation command needs an event type; without one the outbox row reaches no worker"
+              + " and the operation is accepted into silence");
+    }
+  }
+}
@@ -0,0 +1,26 @@
+package dev.caskeleton.application.operation;
+
+import java.util.Objects;
+
+/**
+ * Why a durable operation failed, in terms a transport can publish.
+ *
+ * <p>A code and an already-safe message rather than an exception, because this is stored and later
+ * handed to whoever polls. An exception would carry a stack trace, a SQL fragment, or a host name
+ * into a client-visible document, and the moment it is persisted there is no longer anywhere to
+ * sanitise it.
+ *
+ * @param code the stable failure code the transport maps to its own vocabulary
+ * @param detail a message already safe to publish
+ * @param retryable whether resubmitting the same work could succeed
+ */
+public record OperationFailure(String code, String detail, boolean retryable) {
+
+  public OperationFailure {
+    Objects.requireNonNull(code, "code");
+    Objects.requireNonNull(detail, "detail");
+    if (code.isBlank()) {
+      throw new IllegalArgumentException("a failure needs a code a caller can branch on");
+    }
+  }
+}
@@ -0,0 +1,42 @@
+package dev.caskeleton.application.operation;
+
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * A worker's expiring claim on an operation.
+ *
+ * <p>Expiring is what makes it a lease rather than a flag. A worker can be killed between claiming
+ * an operation and finishing it, and a claim that never lapses leaves the operation RUNNING for
+ * ever with no worker behind it — the failure that makes a queue stop draining while every
+ * dashboard reports it busy.
+ *
+ * @param workerId who holds it
+ * @param acquiredAt when it was taken
+ * @param expiresAt when another worker may take it
+ */
+public record OperationLease(String workerId, Instant acquiredAt, Instant expiresAt) {
+
+  public OperationLease {
+    Objects.requireNonNull(workerId, "workerId");
+    Objects.requireNonNull(acquiredAt, "acquiredAt");
+    Objects.requireNonNull(expiresAt, "expiresAt");
+    if (workerId.isBlank()) {
+      throw new IllegalArgumentException(
+          "a lease needs an owner; an anonymous one cannot be verified when the work completes");
+    }
+    if (!expiresAt.isAfter(acquiredAt)) {
+      throw new IllegalArgumentException("a lease that expires when taken is not a lease");
+    }
+  }
+
+  /** Whether this lease has lapsed. */
+  public boolean expiredAt(Instant now) {
+    return !now.isBefore(expiresAt);
+  }
+
+  /** The same lease, held longer. */
+  public OperationLease extendedTo(Instant newExpiry) {
+    return new OperationLease(workerId, acquiredAt, newExpiry);
+  }
+}
@@ -0,0 +1,35 @@
+package dev.caskeleton.application.operation;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * How far a running operation has got.
+ *
+ * <p>The total is optional because most work honestly cannot predict it. A store that required one
+ * would get an invented number, and clients build progress bars out of invented numbers.
+ *
+ * @param completedUnits work finished so far
+ * @param totalUnits work expected in total, when known
+ * @param phase a label for the current stage, when the work has stages
+ */
+public record OperationProgressSnapshot(
+    long completedUnits, Optional<Long> totalUnits, Optional<String> phase) {
+
+  public OperationProgressSnapshot {
+    Objects.requireNonNull(totalUnits, "totalUnits");
+    Objects.requireNonNull(phase, "phase");
+    if (completedUnits < 0) {
+      throw new IllegalArgumentException("completed units cannot be negative");
+    }
+    if (totalUnits.isPresent() && totalUnits.get() < completedUnits) {
+      throw new IllegalArgumentException(
+          "an operation cannot have completed more units than it has");
+    }
+  }
+
+  /** Progress with no known total. */
+  public static OperationProgressSnapshot of(long completedUnits) {
+    return new OperationProgressSnapshot(completedUnits, Optional.empty(), Optional.empty());
+  }
+}
@@ -0,0 +1,121 @@
+package dev.caskeleton.application.operation;
+
+import dev.caskeleton.application.capability.Idempotency;
+import dev.caskeleton.application.capability.RepositoryAccess;
+import dev.caskeleton.application.capability.UseCaseCapability;
+import dev.caskeleton.application.outbox.NewOutboxEvent;
+import dev.caskeleton.application.outbox.OutboxAppendPort;
+import dev.caskeleton.application.security.RequiresPermission;
+import dev.caskeleton.application.transaction.TransactionMode;
+import dev.caskeleton.application.transaction.TransactionPort;
+import java.time.Clock;
+import java.util.Objects;
+
+/**
+ * Accepts long-running work: one transaction that writes the operation row and the outbox row.
+ *
+ * <p>Both in one transaction, and the 202 only after it commits. The two obvious alternatives are
+ * both broken in ways that only show under failure:
+ *
+ * <ul>
+ *   <li>Publishing to the broker directly, then writing the row — the publish can succeed and the
+ *       row can fail, so a worker starts work that nothing is tracking, and the client is told the
+ *       submission failed.
+ *   <li>Writing the row, committing, then publishing — the commit can succeed and the publish can
+ *       fail, so the operation sits PENDING for ever and nothing ever picks it up.
+ * </ul>
+ *
+ * <p>The outbox exists precisely so neither happens: the row and the intent to publish commit
+ * together or not at all, and the relay makes the publish eventually happen. A duplicate publish
+ * from the relay is fine — it carries the same operation id, so a worker's claim deduplicates it.
+ *
+ * <p>This class does not touch the broker. Acknowledgement, retry and dead-lettering belong to the
+ * messaging platform, and reimplementing any of them here would be a second, weaker copy of a
+ * subsystem this repository already vendors.
+ */
+// Accepting long-running work is a write, so it takes a permission gate like any other. The
+// operation it will eventually run has its own authorization; this one governs who may enqueue.
+@RequiresPermission("operation:submit")
+@UseCaseCapability(
+    transactionMode = TransactionMode.WRITE,
+    // KEYED, not IDEMPOTENT. Re-running this use case is only safe because the caller supplies the
+    // operation id and the fingerprint scope; without a key it would accept the same work twice.
+    idempotency = Idempotency.KEYED,
+    repositoryAccess = RepositoryAccess.WRITE_REPOSITORY,
+    // False, and load-bearing. Nothing here reaches the broker: the outbox row is the publish
+    // intent and the relay performs it. A use case that published directly could not be in one
+    // transaction with the operation row, which is the whole point.
+    externalOutboundAllowed = false)
+// Not final: @RequiresPermission is read by an AOP-proxying authorization manager, and a final
+// class cannot be proxied — so the gate would be declared and never applied.
+// RequiresPermissionProxyabilityTest
+// enforces this, which is the only reason the omission is not a silent one.
+public class SubmitDurableOperationUseCase {
+
+  private final DurableOperationStorePort operations;
+  private final OutboxAppendPort outbox;
+  private final TransactionPort transactions;
+  private final Clock clock;
+
+  /**
+   * A use case over the operation store and the outbox.
+   *
+   * @param operations where operations are recorded
+   * @param outbox where the publish intent is recorded
+   * @param transactions opens the write transaction both share
+   * @param clock the submission clock
+   */
+  public SubmitDurableOperationUseCase(
+      DurableOperationStorePort operations,
+      OutboxAppendPort outbox,
+      TransactionPort transactions,
+      Clock clock) {
+    this.operations = Objects.requireNonNull(operations, "operations");
+    this.outbox = Objects.requireNonNull(outbox, "outbox");
+    this.transactions = Objects.requireNonNull(transactions, "transactions");
+    this.clock = Objects.requireNonNull(clock, "clock");
+  }
+
+  /**
+   * Accepts the command, or returns the operation an identical one already created.
+   *
+   * @param command what to run
+   * @return what to tell the caller
+   */
+  public OperationAcceptance submit(OperationCommand command) {
+    Objects.requireNonNull(command, "command");
+    return transactions.inWrite(
+        () -> {
+          DurableOperationSubmission submission =
+              new DurableOperationSubmission(
+                  command.operationId(),
+                  command.operationName(),
+                  command.principal(),
+                  command.tenantId(),
+                  command.fingerprint(),
+                  command.payload(),
+                  clock.instant(),
+                  command.timeToLive());
+          DurableOperation operation = operations.submit(submission);
+          boolean alreadyAccepted = !operation.operationId().equals(command.operationId());
+
+          if (!alreadyAccepted) {
+            // Appended only for a genuinely new operation. A second outbox row for the same
+            // operation would have a worker claim it, find it already RUNNING or finished, and
+            // discard the message — harmless but indistinguishable in the logs from a real
+            // duplicate, which is exactly the signal an operator needs to stay trustworthy.
+            outbox.append(
+                new NewOutboxEvent(
+                    operation.operationId().value(),
+                    command.eventType(),
+                    operation.operationId().value(),
+                    command.payload(),
+                    operation.submittedAt(),
+                    operation.operationId().value(),
+                    operation.operationId().value()));
+          }
+          return new OperationAcceptance(
+              operation.operationId(), operation.submittedAt(), alreadyAccepted);
+        });
+  }
+}
@@ -0,0 +1,36 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * A pseudonymous handle for whoever is connected.
+ *
+ * <p>A fingerprint and never an identity, and the type exists to make that non-optional. Everything
+ * downstream of this port writes the value into a datastore key, a broker partition key or a metric
+ * tag — places with none of the access controls the application has, and places whose contents end
+ * up in backups, slow logs and incident screenshots. A raw user id or tenant name there is a
+ * disclosure that no later code change can take back.
+ *
+ * <p>The grammar is enforced rather than documented for the same reason: a caller that passes an
+ * email address gets a failure at the boundary instead of a Redis key containing an email address.
+ *
+ * @param value the fingerprint
+ */
+public record ActorFingerprint(String value) {
+
+  private static final Pattern GRAMMAR = Pattern.compile("[a-zA-Z0-9][a-zA-Z0-9._:-]{7,127}");
+
+  public ActorFingerprint {
+    Objects.requireNonNull(value, "value");
+    if (!GRAMMAR.matcher(value).matches()) {
+      throw new IllegalArgumentException(
+          "an actor fingerprint is 8..128 characters of [a-zA-Z0-9._:-] starting alphanumeric; a"
+              + " value outside that is usually a raw identifier, and this one reaches datastore"
+              + " keys and metric tags");
+    }
+    if (value.indexOf('@') >= 0) {
+      throw new IllegalArgumentException("an address is not a fingerprint");
+    }
+  }
+}
@@ -0,0 +1,42 @@
+package dev.caskeleton.application.realtime;
+
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * One node's report that it is holding connections for an actor.
+ *
+ * <p>A report, not a fact. The node that wrote it is the only thing that knows whether the
+ * connections still exist, and by the time anybody reads this the node may be gone. Everything the
+ * registry returns therefore carries {@code observedAt}, and every reader is expected to judge it.
+ *
+ * @param actor whose connections
+ * @param nodeId which process is holding them
+ * @param channel which feed
+ * @param connectionCount how many the node had at that moment
+ * @param observedAt when it looked
+ */
+public record ConnectionRegistration(
+    ActorFingerprint actor,
+    RealtimeNodeId nodeId,
+    RealtimeChannel channel,
+    int connectionCount,
+    Instant observedAt) {
+
+  public ConnectionRegistration {
+    Objects.requireNonNull(actor, "actor");
+    Objects.requireNonNull(nodeId, "nodeId");
+    Objects.requireNonNull(channel, "channel");
+    Objects.requireNonNull(observedAt, "observedAt");
+    if (connectionCount < 0) {
+      throw new IllegalArgumentException("a connection count cannot be negative");
+    }
+  }
+
+  /** Whether this report is older than the caller is willing to act on. */
+  public boolean staleAt(java.time.Duration maximumAge, Instant now) {
+    Objects.requireNonNull(maximumAge, "maximumAge");
+    Objects.requireNonNull(now, "now");
+    return !now.isBefore(observedAt.plus(maximumAge));
+  }
+}
@@ -0,0 +1,57 @@
+package dev.caskeleton.application.realtime;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * Where an actor is currently connected, across every node.
+ *
+ * <p>A cache of state owned by other processes, and the port says so rather than leaving each
+ * caller to discover it. Two expiry mechanisms exist because neither covers the other's case: a TTL
+ * per entry removes what a node forgot to clean up, and a node heartbeat removes everything a node
+ * was holding when it died. A TTL alone leaves a dead node's entries live until each one
+ * individually expires; a heartbeat alone leaves a live node's stale entries for ever.
+ *
+ * <p>Nothing here is authoritative and nothing security-relevant may depend on it. An attacker who
+ * can stop a node reporting can move this registry's answer, so "is this actor connected" is a
+ * question for showing a green dot and for deciding where to route a message that will be dropped
+ * if it is wrong — not for deciding whether a request is legitimate.
+ *
+ * <p>Every implementation is expected to degrade rather than throw. A registry that is unreachable
+ * should report nothing found, because the caller's fallback — broadcast to the cluster and let the
+ * holder claim it — is correct and an exception is not.
+ */
+public interface ConnectionRegistryPort {
+
+  /**
+   * Record that a node holds connections for an actor.
+   *
+   * @param registration what the node observed
+   * @param timeToLive how long the entry survives without being refreshed
+   */
+  void announce(ConnectionRegistration registration, Duration timeToLive);
+
+  /** Remove one node's entry for an actor. */
+  void withdraw(ActorFingerprint actor, RealtimeChannel channel, RealtimeNodeId nodeId);
+
+  /**
+   * Where an actor is reported connected.
+   *
+   * @param now the instant staleness is judged against; entries older than their TTL are omitted
+   * @return one entry per node, newest first; empty when nothing is known
+   */
+  List<ConnectionRegistration> locate(ActorFingerprint actor, RealtimeChannel channel, Instant now);
+
+  /** Record that a node is alive. */
+  void heartbeat(RealtimeNodeId nodeId, Instant now);
+
+  /**
+   * Remove everything held by nodes that have stopped reporting.
+   *
+   * @param heartbeatTimeout how long a node may be silent before it is treated as gone
+   * @param now the current instant
+   * @return the nodes whose entries were removed
+   */
+  List<RealtimeNodeId> evictSilentNodes(Duration heartbeatTimeout, Instant now);
+}
@@ -0,0 +1,23 @@
+package dev.caskeleton.application.realtime;
+
+/**
+ * At-least-once delivery to whichever node holds the recipient.
+ *
+ * <p>At-least-once is the contract, and saying so is the point of the name. A redeploy, a slow
+ * consumer or a broker rebalance all redeliver, so a receiver that assumes exactly-once will show a
+ * user the same thing twice on a perfectly ordinary Tuesday. Deduplication belongs at the receiver,
+ * keyed on the record's stream and position.
+ *
+ * <p>Publication is not delivery. A successful call means the transport accepted the record, which
+ * is a statement about the transport and not about any recipient.
+ */
+public interface DurableFanoutPort {
+
+  /**
+   * Publish a record.
+   *
+   * @param record what to publish
+   * @throws DurableFanoutUnavailableException when the transport would not accept it
+   */
+  void publish(DurableFanoutRecord record);
+}
@@ -0,0 +1,67 @@
+package dev.caskeleton.application.realtime;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * One message published for delivery to whichever node holds its recipient.
+ *
+ * <p>{@code expiresAt} is carried on the record and checked by the receiver, not only by the
+ * publisher. A durable transport is exactly the one that can hold a message through an outage and
+ * deliver it afterwards, and for the traffic this carries — a notification, a live update — late is
+ * indistinguishable from wrong. The receiving node is the only place that knows what time it is
+ * when the message finally arrives.
+ *
+ * <p>The ordering pair is present or absent together. A stream id without a position cannot be
+ * deduplicated and a position without a stream cannot be interpreted, so half of one is a caller
+ * error rather than a partial ordering.
+ *
+ * @param channel which feed
+ * @param partitionKey what the transport should order and route by, usually the recipient
+ * @param payload the encoded message, as a document rather than raw bytes — a byte array cannot be
+ *     a record component without breaking equality, and every producer here is already serialising
+ *     through the message catalog
+ * @param streamId the ordering stream, when the message is ordered
+ * @param position the position within that stream
+ * @param publishedAt when it was published
+ * @param expiresAt after which a receiver must drop it
+ */
+public record DurableFanoutRecord(
+    RealtimeChannel channel,
+    String partitionKey,
+    String payload,
+    Optional<String> streamId,
+    Optional<Long> position,
+    Instant publishedAt,
+    Instant expiresAt) {
+
+  public DurableFanoutRecord {
+    Objects.requireNonNull(channel, "channel");
+    Objects.requireNonNull(partitionKey, "partitionKey");
+    Objects.requireNonNull(payload, "payload");
+    Objects.requireNonNull(streamId, "streamId");
+    Objects.requireNonNull(position, "position");
+    Objects.requireNonNull(publishedAt, "publishedAt");
+    Objects.requireNonNull(expiresAt, "expiresAt");
+    if (partitionKey.isBlank()) {
+      throw new IllegalArgumentException(
+          "a blank partition key sends every message to one partition, which serialises the whole"
+              + " feed behind its slowest recipient");
+    }
+    if (streamId.isPresent() != position.isPresent()) {
+      throw new IllegalArgumentException(
+          "a stream id and a position are meaningless apart: one cannot be deduplicated and the"
+              + " other cannot be interpreted");
+    }
+    if (!expiresAt.isAfter(publishedAt)) {
+      throw new IllegalArgumentException("a record that expires before it is published is dropped");
+    }
+  }
+
+  /** Whether a receiver must drop this rather than deliver it. */
+  public boolean expiredAt(Instant now) {
+    Objects.requireNonNull(now, "now");
+    return !now.isBefore(expiresAt);
+  }
+}
@@ -0,0 +1,30 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.Objects;
+
+/**
+ * The durable transport would not accept a record.
+ *
+ * <p>Distinct from a delivery failure, which this port cannot observe. It means the message was not
+ * published, so the caller still owns it — retrying, dropping it, or falling back to a best-effort
+ * publish are all defensible and the caller is the only one that knows which.
+ *
+ * <p>The channel is named and the payload is not: the payload is the user's data and this exception
+ * reaches logs.
+ */
+public final class DurableFanoutUnavailableException extends RuntimeException {
+
+  private static final long serialVersionUID = 1L;
+
+  private final transient RealtimeChannel channel;
+
+  public DurableFanoutUnavailableException(RealtimeChannel channel, Throwable cause) {
+    super("the durable transport did not accept a record for channel " + channel.value(), cause);
+    this.channel = Objects.requireNonNull(channel, "channel");
+  }
+
+  /** Which channel. */
+  public RealtimeChannel channel() {
+    return channel;
+  }
+}
@@ -0,0 +1,39 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.function.BiConsumer;
+
+/**
+ * Best-effort delivery to every node listening on a channel.
+ *
+ * <p>Ephemeral is the whole contract and it is stated in the name because the alternative is that
+ * somebody assumes otherwise. A message published here reaches the nodes that are subscribed at
+ * that instant and nobody else: a node that is restarting misses it, a subscriber that reconnects a
+ * second later misses it, and nothing is retained, retried or acknowledged. There is no replay.
+ *
+ * <p>That is the right shape for presence updates, typing indicators and cache invalidations —
+ * things whose next update supersedes the one that was lost. It is the wrong shape for anything a
+ * user is told happened, and {@link DurableFanoutPort} exists for those.
+ */
+public interface EphemeralFanoutPort {
+
+  /**
+   * Publish to everyone currently listening.
+   *
+   * @param channel where to publish
+   * @param payload the encoded message
+   * @return how many subscribers the transport reported, which is a diagnostic and not a delivery
+   *     guarantee — the count is what the broker saw, not what any node processed
+   */
+  long publish(RealtimeChannel channel, byte[] payload);
+
+  /**
+   * Listen on a channel.
+   *
+   * @param channel what to listen to
+   * @param listener receives the channel name and the payload; must not block, because it runs on
+   *     the transport's delivery thread and a slow listener stalls every other subscription sharing
+   *     it
+   */
+  FanoutSubscription subscribe(
+      RealtimeChannel channel, BiConsumer<RealtimeChannel, byte[]> listener);
+}
@@ -0,0 +1,19 @@
+package dev.caskeleton.application.realtime;
+
+/**
+ * A live subscription to a channel.
+ *
+ * <p>{@link AutoCloseable} because the subscription holds a connection, and one that is never
+ * closed is a connection that is never returned. A node that opens one per stream and closes none
+ * runs out of Redis connections at whatever its stream churn happens to be, which looks like Redis
+ * being unavailable and is not.
+ */
+public interface FanoutSubscription extends AutoCloseable {
+
+  /** Whether the subscription is still delivering. */
+  boolean active();
+
+  /** Stop delivering and release the connection. Idempotent. */
+  @Override
+  void close();
+}
@@ -0,0 +1,31 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.List;
+
+/**
+ * Retained event history, for a client that reconnects and wants to continue.
+ *
+ * <p>The history lives wherever the messaging platform keeps it, and this port is the only way a
+ * transport adapter reaches it. That boundary is the point: a web or websocket module that kept its
+ * own copy would have its own retention, its own eviction and its own opinion about ordering, and
+ * the two would diverge silently because both would look plausible.
+ *
+ * <p>{@link #window} exists so a caller can distinguish "your cursor is too old" from "there is
+ * nothing new", which {@link #replayAfter} returning an empty list cannot.
+ */
+public interface LiveEventReplayPort {
+
+  /** What the stream still holds. */
+  ReplayWindow window(String streamId);
+
+  /**
+   * Events after a position.
+   *
+   * @param streamId which stream
+   * @param afterPosition the last position the client already has; 0 for the beginning
+   * @param limit the most to return, so one call cannot load a whole stream into memory
+   * @return the events in ascending position order, possibly empty
+   * @throws ReplayCursorUnavailableException when the position is older than what is retained
+   */
+  List<ReplayedEvent> replayAfter(String streamId, long afterPosition, int limit);
+}
@@ -0,0 +1,26 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * A logical channel: which feed a registration, a fan-out or a replay belongs to.
+ *
+ * <p>Kept separate from the transport's own endpoint or destination names so that a change to a URL
+ * is not a change to a Redis key or a broker topic. The two drift on purpose — a path can be
+ * renamed for a client's benefit without invalidating every entry a cluster is holding.
+ *
+ * @param value the channel name
+ */
+public record RealtimeChannel(String value) {
+
+  private static final Pattern GRAMMAR = Pattern.compile("[a-z0-9][a-z0-9._-]{0,63}");
+
+  public RealtimeChannel {
+    Objects.requireNonNull(value, "value");
+    if (!GRAMMAR.matcher(value).matches()) {
+      throw new IllegalArgumentException(
+          "a channel is 1..64 lowercase characters of [a-z0-9._-] starting alphanumeric: " + value);
+    }
+  }
+}
@@ -0,0 +1,26 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * Which process reported something.
+ *
+ * <p>Constrained because it is both a datastore key component and a metric tag. An unconstrained
+ * node id is an unbounded-cardinality tag on every metric that carries it, and in a deployment that
+ * names nodes after pods that is one new tag value per restart.
+ *
+ * @param value the node identity
+ */
+public record RealtimeNodeId(String value) {
+
+  private static final Pattern GRAMMAR = Pattern.compile("[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}");
+
+  public RealtimeNodeId {
+    Objects.requireNonNull(value, "value");
+    if (!GRAMMAR.matcher(value).matches()) {
+      throw new IllegalArgumentException(
+          "a node id is 1..64 characters of [a-zA-Z0-9._-] starting alphanumeric: " + value);
+    }
+  }
+}
@@ -0,0 +1,37 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.Objects;
+
+/**
+ * A client's position is older than the retained history.
+ *
+ * <p>Thrown rather than served from the oldest available position. Serving would deliver a stream
+ * whose positions are contiguous from where the replay started, so the client has no way to see the
+ * hole in the middle — and a client that cannot see a gap does not resnapshot, which is the one
+ * thing that would fix it.
+ *
+ * @see ReplayWindow#canResumeAfter(long)
+ */
+public final class ReplayCursorUnavailableException extends RuntimeException {
+
+  private static final long serialVersionUID = 1L;
+
+  private final transient String streamId;
+  private final transient long requestedPosition;
+
+  public ReplayCursorUnavailableException(String streamId, long requestedPosition) {
+    super("the requested position is older than the retained history; a resnapshot is required");
+    this.streamId = Objects.requireNonNull(streamId, "streamId");
+    this.requestedPosition = requestedPosition;
+  }
+
+  /** Which stream. */
+  public String streamId() {
+    return streamId;
+  }
+
+  /** What was asked for. */
+  public long requestedPosition() {
+    return requestedPosition;
+  }
+}
@@ -0,0 +1,57 @@
+package dev.caskeleton.application.realtime;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * What a stream still holds.
+ *
+ * <p>The type exists so that "can this cursor be honoured" is answerable before anybody tries. A
+ * replay source that only offers "give me everything after N" leaves the caller to infer eviction
+ * from an empty result — and an empty result also means "nothing new", which is the opposite
+ * situation and needs the opposite response.
+ *
+ * @param streamId which stream
+ * @param earliestRetained the oldest position still held, absent when the stream holds nothing
+ * @param latestRetained the newest position held, absent when the stream holds nothing
+ */
+public record ReplayWindow(
+    String streamId, Optional<Long> earliestRetained, Optional<Long> latestRetained) {
+
+  public ReplayWindow {
+    Objects.requireNonNull(streamId, "streamId");
+    Objects.requireNonNull(earliestRetained, "earliestRetained");
+    Objects.requireNonNull(latestRetained, "latestRetained");
+    if (earliestRetained.isPresent() != latestRetained.isPresent()) {
+      throw new IllegalArgumentException(
+          "a window has both bounds or neither; one alone describes no range");
+    }
+    if (earliestRetained.isPresent() && earliestRetained.get() > latestRetained.get()) {
+      throw new IllegalArgumentException("a window cannot start after it ends");
+    }
+  }
+
+  /** A stream that holds nothing. */
+  public static ReplayWindow empty(String streamId) {
+    return new ReplayWindow(streamId, Optional.empty(), Optional.empty());
+  }
+
+  /**
+   * Whether a client asking to continue after {@code position} can be served.
+   *
+   * <p>False for a position older than what is retained, which must produce a resnapshot rather
+   * than a replay from the oldest available: the positions after the hole are contiguous, so
+   * nothing in the delivered data would say events are missing.
+   *
+   * <p>True for a position at or beyond the newest retained — the client is simply up to date, and
+   * that is a live subscription with no replay rather than a failure.
+   */
+  public boolean canResumeAfter(long position) {
+    if (earliestRetained.isEmpty()) {
+      // Nothing retained at all. Only a client that is already current can be served, and only
+      // because there is nothing to send it.
+      return position >= 0;
+    }
+    return position + 1 >= earliestRetained.get();
+  }
+}
@@ -0,0 +1,33 @@
+package dev.caskeleton.application.realtime;
+
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * One retained event, at the position it occupies in its stream.
+ *
+ * <p>Positions count from 1 so that an unset field cannot pass for the first event. A client
+ * resuming from position 0 and one resuming from "wherever the stream starts" are different
+ * requests, and a zero that means both is how a resume silently re-reads history the client already
+ * had.
+ *
+ * @param streamId which stream
+ * @param position where in it
+ * @param payload the encoded event
+ * @param occurredAt when it happened
+ */
+public record ReplayedEvent(String streamId, long position, String payload, Instant occurredAt) {
+
+  public ReplayedEvent {
+    Objects.requireNonNull(streamId, "streamId");
+    Objects.requireNonNull(payload, "payload");
+    Objects.requireNonNull(occurredAt, "occurredAt");
+    if (streamId.isBlank()) {
+      throw new IllegalArgumentException("a stream must be named");
+    }
+    if (position < 1) {
+      throw new IllegalArgumentException(
+          "stream positions count from 1, so an unset field cannot pass for the first event");
+    }
+  }
+}
@@ -0,0 +1,306 @@
+package dev.caskeleton.application.operation;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A store with the semantics the JPA adapter must also have.
+ *
+ * <p>Written to the same rules rather than to whatever makes the tests pass: a submission scope
+ * that already exists returns the existing operation, a lease is checked on every write, and a
+ * terminal operation is never reopened. A fake that was laxer than the adapter would let the use
+ * case tests certify behaviour the real store refuses.
+ */
+final class InMemoryDurableOperationStore implements DurableOperationStorePort {
+
+  private record Scope(String tenant, String principal, String operationName, String hash) {}
+
+  private final Map<DurableOperationId, DurableOperation> byId = new ConcurrentHashMap<>();
+  private final Map<Scope, DurableOperationId> byScope = new ConcurrentHashMap<>();
+  private final Map<DurableOperationId, String> payloads = new ConcurrentHashMap<>();
+
+  @Override
+  public DurableOperation submit(DurableOperationSubmission submission) {
+    Scope scope =
+        new Scope(
+            submission.tenantId() == null ? "" : submission.tenantId(),
+            submission.principal(),
+            submission.operationName(),
+            submission.fingerprint().hex());
+    DurableOperationId existing = byScope.putIfAbsent(scope, submission.operationId());
+    if (existing != null) {
+      return byId.get(existing);
+    }
+    DurableOperation operation =
+        new DurableOperation(
+            submission.operationId(),
+            submission.operationName(),
+            submission.principal(),
+            submission.tenantId(),
+            DurableOperationState.PENDING,
+            submission.submittedAt(),
+            Optional.empty(),
+            Optional.empty(),
+            Optional.empty(),
+            Optional.empty(),
+            Optional.empty(),
+            Optional.empty(),
+            submission.expiresAt());
+    byId.put(submission.operationId(), operation);
+    payloads.put(submission.operationId(), submission.payload());
+    return operation;
+  }
+
+  @Override
+  public Optional<DurableOperation> find(
+      DurableOperationId operationId, String principal, Instant now) {
+    return Optional.ofNullable(byId.get(operationId))
+        .filter(operation -> operation.principal().equals(principal));
+  }
+
+  @Override
+  public Optional<DurableOperationSubmission> claimNext(
+      String workerId, Duration leaseDuration, Instant now) {
+    return byId.values().stream()
+        .filter(
+            operation ->
+                operation.state() == DurableOperationState.PENDING
+                    || (operation.state() == DurableOperationState.RUNNING
+                        && operation.leaseExpiredAt(now)))
+        .filter(operation -> now.isBefore(operation.expiresAt()))
+        .min(Comparator.comparing(DurableOperation::submittedAt))
+        .map(
+            operation -> {
+              DurableOperation claimed =
+                  replace(
+                      operation,
+                      DurableOperationState.RUNNING,
+                      Optional.of(operation.startedAt().orElse(now)),
+                      Optional.empty(),
+                      operation.progress(),
+                      operation.resultReference(),
+                      Optional.empty(),
+                      Optional.of(new OperationLease(workerId, now, now.plus(leaseDuration))));
+              byId.put(claimed.operationId(), claimed);
+              return new DurableOperationSubmission(
+                  claimed.operationId(),
+                  claimed.operationName(),
+                  claimed.principal(),
+                  claimed.tenantId(),
+                  new dev.caskeleton.application.idempotency.RequestFingerprint("0".repeat(64)),
+                  payloads.getOrDefault(claimed.operationId(), "{}"),
+                  claimed.submittedAt(),
+                  Duration.between(claimed.submittedAt(), claimed.expiresAt()));
+            });
+  }
+
+  @Override
+  public boolean heartbeat(
+      DurableOperationId operationId, String workerId, Duration leaseDuration, Instant now) {
+    DurableOperation operation = byId.get(operationId);
+    if (!holdsLease(operation, workerId, now)) {
+      return false;
+    }
+    byId.put(
+        operationId,
+        replace(
+            operation,
+            operation.state(),
+            operation.startedAt(),
+            operation.completedAt(),
+            operation.progress(),
+            operation.resultReference(),
+            Optional.empty(),
+            operation.lease().map(lease -> lease.extendedTo(now.plus(leaseDuration)))));
+    return true;
+  }
+
+  @Override
+  public boolean reportProgress(
+      DurableOperationId operationId, String workerId, OperationProgressSnapshot progress) {
+    DurableOperation operation = byId.get(operationId);
+    if (operation == null || !ownedBy(operation, workerId)) {
+      return false;
+    }
+    byId.put(
+        operationId,
+        replace(
+            operation,
+            operation.state(),
+            operation.startedAt(),
+            operation.completedAt(),
+            Optional.of(progress),
+            operation.resultReference(),
+            Optional.empty(),
+            operation.lease()));
+    return true;
+  }
+
+  @Override
+  public boolean succeed(
+      DurableOperationId operationId,
+      String workerId,
+      String resultReference,
+      Instant completedAt) {
+    DurableOperation operation = byId.get(operationId);
+    if (operation == null || !ownedBy(operation, workerId)) {
+      return false;
+    }
+    byId.put(
+        operationId,
+        replace(
+            operation,
+            DurableOperationState.SUCCEEDED,
+            operation.startedAt(),
+            Optional.of(completedAt),
+            operation.progress(),
+            Optional.of(resultReference),
+            Optional.empty(),
+            Optional.empty()));
+    return true;
+  }
+
+  @Override
+  public boolean fail(
+      DurableOperationId operationId,
+      String workerId,
+      OperationFailure failure,
+      Instant completedAt) {
+    DurableOperation operation = byId.get(operationId);
+    if (operation == null || !ownedBy(operation, workerId)) {
+      return false;
+    }
+    byId.put(
+        operationId,
+        replace(
+            operation,
+            DurableOperationState.FAILED,
+            operation.startedAt(),
+            Optional.of(completedAt),
+            operation.progress(),
+            Optional.empty(),
+            Optional.of(failure),
+            Optional.empty()));
+    return true;
+  }
+
+  @Override
+  public boolean cancel(DurableOperationId operationId, String principal, Instant canceledAt) {
+    DurableOperation operation = byId.get(operationId);
+    if (operation == null
+        || !operation.principal().equals(principal)
+        || !operation.state().cancellable()) {
+      return false;
+    }
+    byId.put(
+        operationId,
+        replace(
+            operation,
+            DurableOperationState.CANCELED,
+            operation.startedAt(),
+            Optional.of(canceledAt),
+            operation.progress(),
+            operation.resultReference(),
+            Optional.empty(),
+            Optional.empty()));
+    return true;
+  }
+
+  @Override
+  public List<DurableOperationId> reclaimExpiredLeases(Instant now) {
+    List<DurableOperationId> reclaimed = new ArrayList<>();
+    byId.forEach(
+        (id, operation) -> {
+          if (operation.state() == DurableOperationState.RUNNING && operation.leaseExpiredAt(now)) {
+            byId.put(
+                id,
+                replace(
+                    operation,
+                    DurableOperationState.PENDING,
+                    operation.startedAt(),
+                    Optional.empty(),
+                    operation.progress(),
+                    operation.resultReference(),
+                    Optional.empty(),
+                    Optional.empty()));
+            reclaimed.add(id);
+          }
+        });
+    return List.copyOf(reclaimed);
+  }
+
+  @Override
+  public List<DurableOperationId> expireStaleOperations(Instant now) {
+    List<DurableOperationId> expired = new ArrayList<>();
+    byId.forEach(
+        (id, operation) -> {
+          if (!operation.state().terminal() && !now.isBefore(operation.expiresAt())) {
+            byId.put(
+                id,
+                replace(
+                    operation,
+                    DurableOperationState.EXPIRED,
+                    operation.startedAt(),
+                    Optional.of(now),
+                    operation.progress(),
+                    operation.resultReference(),
+                    Optional.empty(),
+                    Optional.empty()));
+            expired.add(id);
+          }
+        });
+    return List.copyOf(expired);
+  }
+
+  /** How many operations exist. */
+  int size() {
+    return byId.size();
+  }
+
+  /** Drops everything, standing in for a rolled-back transaction. */
+  void clear() {
+    byId.clear();
+    byScope.clear();
+    payloads.clear();
+  }
+
+  private static boolean holdsLease(DurableOperation operation, String workerId, Instant now) {
+    return operation != null && ownedBy(operation, workerId) && !operation.leaseExpiredAt(now);
+  }
+
+  private static boolean ownedBy(DurableOperation operation, String workerId) {
+    return operation.state() == DurableOperationState.RUNNING
+        && operation.lease().map(lease -> lease.workerId().equals(workerId)).orElse(false);
+  }
+
+  private static DurableOperation replace(
+      DurableOperation operation,
+      DurableOperationState state,
+      Optional<Instant> startedAt,
+      Optional<Instant> completedAt,
+      Optional<OperationProgressSnapshot> progress,
+      Optional<String> resultReference,
+      Optional<OperationFailure> failure,
+      Optional<OperationLease> lease) {
+    return new DurableOperation(
+        operation.operationId(),
+        operation.operationName(),
+        operation.principal(),
+        operation.tenantId(),
+        state,
+        operation.submittedAt(),
+        startedAt,
+        completedAt,
+        progress,
+        resultReference,
+        failure,
+        lease,
+        operation.expiresAt());
+  }
+}
@@ -0,0 +1,236 @@
+package dev.caskeleton.application.operation;
+
+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.outbox.NewOutboxEvent;
+import dev.caskeleton.application.outbox.OutboxAppendPort;
+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.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The two writes that must not come apart.
+ *
+ * <p>Accepting long-running work means recording the operation and recording the intent to publish
+ * it. Either one alone is a specific outage: a published message with no row is work nobody is
+ * tracking, and a row with no message is an operation that sits PENDING for ever. These cases are
+ * each one way the pair could split.
+ */
+class SubmitDurableOperationUseCaseTest {
+
+  private static final Instant NOW = Instant.parse("2026-08-25T10:00:00Z");
+  private static final DurableOperationId ID = new DurableOperationId("op-1");
+
+  private final InMemoryDurableOperationStore operations = new InMemoryDurableOperationStore();
+  private final AtomicBoolean insideTransaction = new AtomicBoolean();
+  private final RecordingOutbox outbox = new RecordingOutbox(insideTransaction);
+  private final RollbackAwareTransactions transactions =
+      new RollbackAwareTransactions(operations, outbox, insideTransaction);
+  private final SubmitDurableOperationUseCase useCase =
+      new SubmitDurableOperationUseCase(
+          operations, outbox, transactions, Clock.fixed(NOW, ZoneOffset.UTC));
+
+  private static OperationCommand command(DurableOperationId id, String hash) {
+    return new OperationCommand(
+        id,
+        "reports.generate",
+        "alice",
+        null,
+        new RequestFingerprint(hash),
+        "{\"range\":\"2026-08\"}",
+        "ReportRequested",
+        Duration.ofHours(6));
+  }
+
+  @Test
+  @DisplayName("an accepted submission writes both the operation and the outbox event")
+  void submissionWritesBoth() {
+    OperationAcceptance acceptance = useCase.submit(command(ID, "a".repeat(64)));
+
+    assertThat(acceptance.operationId()).isEqualTo(ID);
+    assertThat(acceptance.alreadyAccepted()).isFalse();
+    assertThat(operations.size()).isEqualTo(1);
+    assertThat(outbox.appended).hasSize(1);
+  }
+
+  @Test
+  @DisplayName("both writes happen inside one transaction")
+  void bothWritesShareOneTransaction() {
+    useCase.submit(command(ID, "a".repeat(64)));
+
+    // Not an implementation detail. Outside one transaction the two writes can commit
+    // independently, and every failure mode this use case exists to prevent comes back.
+    assertThat(outbox.appendedOutsideTransaction).isEmpty();
+    assertThat(transactions.independentCommits).isEmpty();
+    assertThat(transactions.writeTransactions).isEqualTo(1);
+  }
+
+  @Test
+  @DisplayName("a rollback leaves neither an operation nor an outbox event")
+  void rollbackLeavesNothing() {
+    outbox.failNext = true;
+
+    assertThatThrownBy(() -> useCase.submit(command(ID, "a".repeat(64))))
+        .isInstanceOf(IllegalStateException.class);
+
+    assertThat(operations.size()).isZero();
+    assertThat(outbox.appended).isEmpty();
+  }
+
+  @Test
+  @DisplayName("an identical resubmission returns the first operation and publishes nothing new")
+  void resubmissionReusesTheFirstOperation() {
+    useCase.submit(command(ID, "a".repeat(64)));
+
+    OperationAcceptance repeat =
+        useCase.submit(command(new DurableOperationId("op-2"), "a".repeat(64)));
+
+    // The client is pointed at the work that is actually running. Starting a second operation
+    // would run the same report twice and hand back an id for the copy.
+    assertThat(repeat.operationId()).isEqualTo(ID);
+    assertThat(repeat.alreadyAccepted()).isTrue();
+    assertThat(operations.size()).isEqualTo(1);
+    assertThat(outbox.appended).hasSize(1);
+  }
+
+  @Test
+  @DisplayName("a different request under the same principal is a different operation")
+  void differentRequestsAreDifferentOperations() {
+    useCase.submit(command(ID, "a".repeat(64)));
+
+    OperationAcceptance second =
+        useCase.submit(command(new DurableOperationId("op-2"), "b".repeat(64)));
+
+    assertThat(second.alreadyAccepted()).isFalse();
+    assertThat(operations.size()).isEqualTo(2);
+    assertThat(outbox.appended).hasSize(2);
+  }
+
+  @Test
+  @DisplayName("the outbox event carries the operation id as its routing key and dedup key")
+  void outboxEventIsAnchoredToTheOperation() {
+    useCase.submit(command(ID, "a".repeat(64)));
+
+    NewOutboxEvent event = outbox.appended.get(0);
+
+    // A worker that receives the message twice — which the relay's at-least-once delivery makes
+    // ordinary — must be able to tell it is the same work. The operation id is what says so.
+    assertThat(event.aggregateId()).isEqualTo("op-1");
+    assertThat(event.idempotencyKey()).isEqualTo("op-1");
+    assertThat(event.eventType()).isEqualTo("ReportRequested");
+  }
+
+  @Test
+  @DisplayName("a command with no event type is refused rather than accepted into silence")
+  void missingEventTypeIsRefused() {
+    assertThatThrownBy(
+            () ->
+                new OperationCommand(
+                    ID,
+                    "reports.generate",
+                    "alice",
+                    null,
+                    new RequestFingerprint("a".repeat(64)),
+                    "{}",
+                    "  ",
+                    Duration.ofHours(1)))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("reaches no worker");
+  }
+
+  private static final class RecordingOutbox implements OutboxAppendPort {
+
+    private final List<NewOutboxEvent> appended = new ArrayList<>();
+    private final List<String> appendedOutsideTransaction = new ArrayList<>();
+    private final AtomicBoolean insideTransaction;
+    private boolean failNext;
+
+    private RecordingOutbox(AtomicBoolean insideTransaction) {
+      this.insideTransaction = insideTransaction;
+    }
+
+    @Override
+    public void append(NewOutboxEvent event) {
+      if (!insideTransaction.get()) {
+        // The port's own contract: appending outside the caller's write transaction is silent
+        // event loss, and it is invisible to any assertion about what ended up in the list.
+        appendedOutsideTransaction.add(event.eventId());
+      }
+      if (failNext) {
+        failNext = false;
+        throw new IllegalStateException("the outbox write failed");
+      }
+      appended.add(event);
+    }
+  }
+
+  /**
+   * A transaction port that actually rolls back.
+   *
+   * <p>A port that simply runs the action would let {@link #rollbackLeavesNothing} pass for an
+   * implementation with no transaction at all — the exact defect the test is looking for.
+   */
+  private static final class RollbackAwareTransactions implements TransactionPort {
+
+    private final InMemoryDurableOperationStore operations;
+    private final RecordingOutbox outbox;
+    private final AtomicBoolean insideTransaction;
+    private final List<String> independentCommits = new ArrayList<>();
+    private int writeTransactions;
+
+    private RollbackAwareTransactions(
+        InMemoryDurableOperationStore operations,
+        RecordingOutbox outbox,
+        AtomicBoolean insideTransaction) {
+      this.operations = operations;
+      this.outbox = outbox;
+      this.insideTransaction = insideTransaction;
+    }
+
+    @Override
+    public <T> T inWrite(Supplier<T> action) {
+      writeTransactions++;
+      insideTransaction.set(true);
+      int outboxMark = outbox.appended.size();
+      try {
+        return action.get();
+      } catch (RuntimeException rollback) {
+        operations.clear();
+        while (outbox.appended.size() > outboxMark) {
+          outbox.appended.remove(outbox.appended.size() - 1);
+        }
+        throw rollback;
+      } finally {
+        insideTransaction.set(false);
+      }
+    }
+
+    @Override
+    public <T> T inRootWrite(Supplier<T> action) {
+      return inWrite(action);
+    }
+
+    @Override
+    public <T> T inRead(Supplier<T> action) {
+      return action.get();
+    }
+
+    @Override
+    public <T> T inNew(Supplier<T> action) {
+      // Deliberately not delegating to inWrite: a REQUIRES_NEW transaction commits independently,
+      // which is the one thing this submission must never do.
+      independentCommits.add("inNew");
+      return action.get();
+    }
+  }
+}