lykakis · Nov 5, 2026 · Sep 28, 2026 · Sep 28, 2026 · Sep 28, 2026 · Sep 28, 2026
diff --git a/docs/14-outbound-storage.md b/docs/14-outbound-storage.md
diff --git a/docs/15-pending-record-schema.md b/docs/15-pending-record-schema.md
diff --git a/docs/DocumentationMap.md b/docs/DocumentationMap.md
@@ -23,6 +23,7 @@ Sendium is an open-source, headless SMS gateway for high-throughput messaging. I
 | Review current features and roadmap | [12. Features And Roadmap](12-features-roadmap.md) |
 | Configure PostgreSQL DLR persistence and review durability | [13. DLR Persistence](13-dlr-persistence.md) |
 | Understand outbound storage and extension contracts | [14. Outbound Message Storage](14-outbound-storage.md) |
+| Implement pending-source encoding and reconstruction | [15. Pending Record Schema](15-pending-record-schema.md) |
 | Contribute code or docs | [Contributing](../.github/CONTRIBUTING.md) |
 
 ## Core Concepts
@@ -65,6 +66,7 @@ Sendium expects these files in the configured `conf` directory. The Docker quick
 | [12. Features And Roadmap](12-features-roadmap.md) | Current product capabilities, planned roadmap phases, and related feature documentation. |
 | [13. DLR Persistence](13-dlr-persistence.md) | PostgreSQL setup, retention, restart guarantees, and durability limits. |
 | [14. Outbound Message Storage](14-outbound-storage.md) | Diagram-led lifecycle guide, class responsibilities, capacity bounds, retries, shutdown, profiles, and extension contracts. |
+| [15. Pending Record Schema](15-pending-record-schema.md) | Versioned CBOR source fields, multipart membership/deadlines, supported extensions, validation and record-driven reconstruction boundaries. |
 | [Kannel migration converter](https://cytechmobile.github.io/sendium/kannelconverter/) | Browser-only helper for turning a legacy `kannel.conf` into Sendium starter files. |
 
 ## API Discovery
@@ -77,7 +79,7 @@ When Sendium is running, the HTTP API can be inspected through:
 | `/swagger-ui` | Interactive Swagger UI. |
 | `/openapi.json` | OpenAPI specification. |
 | `/q/metrics` | Prometheus-compatible Micrometer metrics endpoint. |
-| `/q/health/ready` | Readiness status, including PostgreSQL DLR storage availability. |
+| `/q/health/ready` | Readiness status, including configured DLR storage and the standalone outbound pipeline. |
 
 ## Community And Project Files
 

diff --git a/sendium-app/src/test/java/gr/cytech/sendium/app/storage/MessageLifecycleHttpTest.java b/sendium-app/src/test/java/gr/cytech/sendium/app/storage/MessageLifecycleHttpTest.java
@@ -7,7 +7,11 @@
 import gr.cytech.sendium.core.message.StandardMessage;
 import gr.cytech.sendium.core.outbound.DefaultOutboundCoordinator;
 import gr.cytech.sendium.core.outbound.OutboundCoordinator;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
 import gr.cytech.sendium.core.queue.InMemoryQueueProvider;
+import gr.cytech.sendium.core.storage.OutboundStage;
+import gr.cytech.sendium.core.storage.OutboundStorageException;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
 import gr.cytech.sendium.core.storage.memory.MemoryPendingMessageStore;
 import gr.cytech.sendium.core.storage.memory.MemoryRoutedWorkStore;
 import gr.cytech.sendium.core.storage.memory.MemorySelectedRouterStore;
@@ -19,7 +23,10 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.net.URI;
 import java.net.http.HttpClient;
@@ -32,6 +39,11 @@
 import java.util.concurrent.CompletableFuture;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.spy;
 
 class MessageLifecycleHttpTest {
     @RegisterExtension
@@ -52,14 +64,25 @@ class MessageLifecycleHttpTest {
     @Inject
     InMemoryQueueProvider legacy;
 
+    @Inject
+    PendingMessageStore<StandardMessage> pending;
+
+    @AfterEach
+    void clearAdmissionFaults() {
+        reset(pending);
+    }
+
     @Test
-    void httpUsesTheExplicitCoordinatorBeanAndReturnsBackpressureAtCapacity() throws Exception {
+    void successReferencesAcceptedHandle() throws Exception {
         URI requestUri = URI.create(endpoint + "?username=http-user&password=secret&from=Sender&to=306900000001&text=Hello" +
                 "&dlr-url=https%3A%2F%2Fcallback.test%2Fdlr");
         try (HttpClient client = HttpClient.newHttpClient()) {
             var request = HttpRequest.newBuilder(requestUri).timeout(Duration.ofSeconds(5)).GET().build();
             var accepted = client.send(request, HttpResponse.BodyHandlers.ofString());
             assertThat(accepted.statusCode()).isEqualTo(202);
+            var retained = pending.find(new SourceId(UUID.fromString(accepted.body()))).orElseThrow();
+            assertThat(retained.serial).isEqualTo(accepted.body());
+            assertThat(retained.body).isEqualTo("Hello");
             var full = client.send(request, HttpResponse.BodyHandlers.ofString());
             assertThat(full.statusCode()).isEqualTo(503);
             assertThat(legacy.getRouterQueue().isEmpty()).isTrue();
@@ -75,12 +98,54 @@ void httpUsesTheExplicitCoordinatorBeanAndReturnsBackpressureAtCapacity() throws
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"UNAVAILABLE", "CAPACITY_EXCEEDED", "OWNERSHIP_CONFLICT", "UNSUPPORTED", "INVALID_TRANSITION"})
+    void rejectsPendingAdmissionFailures(String reasonName) throws Exception {
+        // Strings cross QuarkusExtensionTest's class-loader boundary; application enum instances do not.
+        var reason = OutboundStorageException.Reason.valueOf(reasonName);
+        doThrow(new OutboundStorageException(OutboundStage.Role.PENDING, reason, "private storage details"))
+                .when(pending).admit(any(), any());
+        try (HttpClient client = HttpClient.newHttpClient()) {
+            var request = HttpRequest.newBuilder(URI.create(endpoint +
+                    "?username=http-user&password=secret&from=Sender&to=306900000001&text=Hello"))
+                    .timeout(Duration.ofSeconds(5)).GET().build();
+            var response = client.send(request, HttpResponse.BodyHandlers.ofString());
+            int expected = switch (reason) {
+                case UNSUPPORTED -> 400;
+                case INVALID_TRANSITION -> 500;
+                default -> 503;
+            };
+            assertThat(response.statusCode()).isEqualTo(expected);
+            assertThat(response.body()).doesNotContain("private storage details");
+            assertThat(pending.sourcesAfter(null, 1)).isEmpty();
+            assertThat(legacy.getRouterQueue().isEmpty()).isTrue();
+        }
+    }
+
+    @Test
+    void unconfirmedPendingHandleIsNeverAcknowledged() throws Exception {
+        doReturn(null).when(pending).admit(any(), any());
+        try (HttpClient client = HttpClient.newHttpClient()) {
+            var request = HttpRequest.newBuilder(URI.create(endpoint +
+                    "?username=http-user&password=secret&from=Sender&to=306900000001&text=Hello"))
+                    .timeout(Duration.ofSeconds(5)).GET().build();
+            assertThat(client.send(request, HttpResponse.BodyHandlers.ofString()).statusCode()).isEqualTo(503);
+            assertThat(pending.sourcesAfter(null, 1)).isEmpty();
+            assertThat(legacy.getRouterQueue().isEmpty()).isTrue();
+        }
+    }
+
     @Singleton
     public static class Beans {
         @Produces
         @Singleton
-        OutboundCoordinator<StandardMessage> coordinator() {
-            var pending = new MemoryPendingMessageStore<StandardMessage>(1, Beans::snapshot);
+        MemoryPendingMessageStore<StandardMessage> pending() {
+            return spy(new MemoryPendingMessageStore<StandardMessage>(1, Beans::snapshot));
+        }
+
+        @Produces
+        @Singleton
+        OutboundCoordinator<StandardMessage> coordinator(MemoryPendingMessageStore<StandardMessage> pending) {
             var router = new MemorySelectedRouterStore<>(pending, 1);
             var routed = new MemoryRoutedWorkStore<StandardMessage>(1, Beans::snapshot);
             var coordinator = new DefaultOutboundCoordinator<>(pending, router, routed);

diff --git a/sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneMessageSnapshotTest.java b/sendium-app/src/test/java/gr/cytech/sendium/app/storage/StandaloneMessageSnapshotTest.java
@@ -1,16 +1,73 @@
 package gr.cytech.sendium.app.storage;
 
+import gr.cytech.sendium.core.message.DlrReturnMetadata;
 import gr.cytech.sendium.core.message.StandardMessage;
+import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
+import gr.cytech.sendium.core.storage.PendingMessageStore;
+import gr.cytech.sendium.core.storage.memory.MemoryPendingMessageStore;
+import gr.cytech.sendium.core.storage.pending.PendingMessageData;
+import gr.cytech.sendium.core.storage.pending.PendingRecordCodec;
+import gr.cytech.sendium.core.storage.pending.PendingSourceRecord;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class StandaloneMessageSnapshotTest {
+    @Test
+    void normalProtocolRecordPreservesProcessingInputs() {
+        for (var metadata : List.of(DlrReturnMetadata.http("account", "sender", "recipient", "https://callback.test/dlr"),
+                DlrReturnMetadata.smpp("account", "system", "sender", "recipient"))) {
+            var message = new StandardMessage();
+            SourceId source = new SourceId(UUID.randomUUID());
+            message.serial = source.value().toString();
+            message.owner_id = "account";
+            message.systemId = "system";
+            message.from = "sender";
+            message.to = "recipient";
+            message.body = "Hello";
+            message.priority = 3;
+            message.ttl = 120;
+            message.ddt = 5;
+            message.dcs = StandardMessage.DCS_16BIT;
+            message.nextTarget = "provider";
+            message.field1 = 7;
+            message.field2 = 8;
+            message.field3 = 1;
+            message.field4 = "billing-info";
+            message.acked = true;
+            message.dlrReturnMetadata = metadata;
+            message.tlvs = new HashMap<>(Map.of("0427", "1"));
+            message.attrs = new HashMap<>(Map.of("request", "original"));
+
+            try (PendingMessageStore<StandardMessage> pending = new MemoryPendingMessageStore<>(1, StandaloneMessageSnapshot::copy)) {
+                pending.open();
+                SourceId accepted = pending.admit(source, message);
+                var retained = pending.find(accepted).orElseThrow();
+                message.field4 = "changed after admission";
+                message.putAttr("routing-only", "not accepted input");
+                message.tlvs.clear();
+                var codec = new PendingRecordCodec();
+                var record = new PendingSourceRecord(accepted, PendingSourceRecord.Disposition.READY, PendingMessageData.capture(retained));
+                StandardMessage restored = codec.decode(codec.encode(record)).message().restore();
+                assertThat(restored).usingRecursiveComparison().isEqualTo(retained);
+                assertThat(restored.field1).isEqualTo(7);
+                assertThat(restored.field2).isEqualTo(8);
+                assertThat(restored.field3).isEqualTo(1);
+                assertThat(restored.field4).isEqualTo("billing-info");
+                assertThat(restored.dlrReturnMetadata).isEqualTo(metadata);
+                assertThat(restored.tlvs).containsEntry("0427", "1");
+                assertThat(restored.attrs).containsExactlyEntriesOf(Map.of("request", "original"));
+            }
+        }
+    }
+
     @Test
     void snapshotDetachesProtocolCollectionsAndMutableAttributeValues() {
         var message = new StandardMessage();

diff --git a/sendium-core/pom.xml b/sendium-core/pom.xml
@@ -70,6 +70,10 @@
             <groupId>com.fasterxml.jackson.dataformat</groupId>
             <artifactId>jackson-dataformat-yaml</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-cbor</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-junit</artifactId>

diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/http/KannelResource.java b/sendium-core/src/main/java/gr/cytech/sendium/core/http/KannelResource.java
@@ -217,17 +217,20 @@ public Response receiveSms(
             if (msg.acked) {
                 msg.dlrReturnMetadata = DlrReturnMetadata.http(usr, msg.from, msg.to, dlrUrl);
             }
+            String acceptedMessageId;
             if (outboundCoordinators.isUnsatisfied()) {
                 queueProvider.getRouterQueue().enqueue(msg);
+                acceptedMessageId = msg.serial;
             } else {
-                outboundCoordinators.get().accept(new SourceId(gatewayMessageId), msg);
+                SourceId accepted = outboundCoordinators.get().accept(new SourceId(gatewayMessageId), msg);
+                acceptedMessageId = accepted.value().toString();
             }
             if (MessageTrace.shouldLog(configurationHandler, MessageTrace.EVENT_ACCEPTED)) {
                 logger.info("message.accepted ingress=http {}", MessageTrace.identifiers(msg));
             }
 
             return Response.status(Response.Status.ACCEPTED)
-                    .entity(msg.serial)
+                    .entity(acceptedMessageId)
                     .build();
 
         } catch (OutboundStorageException e) {

diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/DefaultOutboundCoordinator.java b/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/DefaultOutboundCoordinator.java
@@ -15,6 +15,7 @@
 import gr.cytech.sendium.core.storage.PendingMessageStore;
 import gr.cytech.sendium.core.storage.RoutedWorkStore;
 import gr.cytech.sendium.core.storage.SelectedRouterStore;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -122,32 +123,58 @@ public void beginShutdown() {
     }
 
     @Override
-    public void accept(SourceId source, M message) {
-        accept(source, message, false);
+    public SourceId accept(SourceId source, M message) {
+        return accept(source, message, false);
     }
 
-    private void accept(SourceId source, M message, boolean held) {
+    private SourceId accept(SourceId source, M message, boolean held) {
         Objects.requireNonNull(source, "source");
         Objects.requireNonNull(message, "message");
         lock.lock();
         try {
             requireRunning(Role.PENDING);
             if (sourceSelections.containsKey(source)) {
-                return;
+                return source;
             }
-            if (held) {
-                pending.admitHeld(source, message);
-            } else {
-                pending.admit(source, message);
+            SourceId accepted = held ? pending.admitHeld(source, message) : pending.admit(source, message);
+            if (!source.equals(accepted)) {
+                throw new OutboundStorageException(Role.PENDING, OutboundStorageException.Reason.OWNERSHIP_CONFLICT,
+                        "Pending admission did not confirm the requested source handle");
             }
+            return accepted;
         } finally {
             lock.unlock();
         }
     }
 
     @Override
-    public void acceptHeld(SourceId source, M message) {
-        accept(source, message, true);
+    public SourceId acceptHeld(SourceId source, M message) {
+        return accept(source, message, true);
+    }
+
+    @Override
+    public MultipartMembership acceptPart(SourceId source, M message, Duration timeout) {
+        Objects.requireNonNull(source, "source");
+        Objects.requireNonNull(message, "message");
+        Objects.requireNonNull(timeout, "timeout");
+        lock.lock();
+        try {
+            requireRunning(Role.PENDING);
+            MultipartMembership accepted;
+            if (sourceSelections.containsKey(source)) {
+                accepted = pending.findMultipart(source)
+                        .orElseThrow(() -> invalid(Role.PENDING, "Multipart source is no longer retained"));
+            } else {
+                accepted = pending.admitPart(source, message, timeout);
+            }
+            if (accepted == null || !source.equals(accepted.source())) {
+                throw new OutboundStorageException(Role.PENDING, OutboundStorageException.Reason.OWNERSHIP_CONFLICT,
+                        "Pending admission did not confirm multipart source membership");
+            }
+            return accepted;
+        } finally {
+            lock.unlock();
+        }
     }
 
     @Override

diff --git a/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java b/sendium-core/src/main/java/gr/cytech/sendium/core/outbound/OutboundCoordinator.java
@@ -7,6 +7,7 @@
 import gr.cytech.sendium.core.outbound.OutboundWork.SelectionId;
 import gr.cytech.sendium.core.outbound.OutboundWork.SourceId;
 import gr.cytech.sendium.core.outbound.OutboundWork.WorkId;
+import gr.cytech.sendium.core.storage.pending.MultipartMembership;
 
 import java.time.Duration;
 import java.util.Optional;
@@ -35,11 +36,14 @@ public interface OutboundCoordinator<M extends StandardMessage> extends AutoClos
      */
     void beginShutdown();
 
-    /** Accepts source ownership and makes the message ready for selection according to the store policy. */
-    void accept(SourceId source, M message);
+    /** Accepts ready source ownership and returns the same confirmed handle before protocol success; null is never acceptance. */
+    SourceId accept(SourceId source, M message);
 
-    /** Accepts source ownership without making this source routable yet. */
-    void acceptHeld(SourceId source, M message);
+    /** Accepts ownership without making the source routable; returns the same confirmed handle as accept. */
+    SourceId acceptHeld(SourceId source, M message);
+
+    /** Accepts a multipart source with stable group membership before protocol success. */
+    MultipartMembership acceptPart(SourceId source, M message, Duration timeout);
 
     /**
      * Makes already-accepted held sources available to bounded selection using one prepared execution message.