refactor(pki): make status authority transactional

Persist immutable status records and STATUS_OBJECT_RECORD content-owner
edges atomically through the transactional metadata store.

Remove the former status/by-id file authority and scan-derived content
retention while keeping CRL payloads external and publication as a
separate post-commit lifecycle.

pki/src/main/java/zeroecho/pki/impl/fs/FilesystemPkiStore.java

@@ -169,8 +169,13 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
     /* package */ static final String CURRENT_STORE_VERSION = "v3";
     private static final String SIGN_RECORD_NAMESPACE = "io.zeroecho.pki.signing-record";
     private static final String SIGN_OWNER_NAMESPACE = "io.zeroecho.pki.signing-owner";
+    private static final String STATUS_RECORD_NAMESPACE = "io.zeroecho.pki.status-object-record";
+    private static final String STATUS_OWNER_NAMESPACE = "io.zeroecho.pki.status-object-owner";
     private static final int CURRENT_SIGN_RECORD_VERSION = 2;
     private static final int SIGN_OWNER_VALUE_VERSION = 1;
+    private static final int STATUS_OWNER_VALUE_VERSION = 1;
+    private static final int METADATA_TRANSFER_BUFFER_BYTES = 16 * 1024;
+    private static final ThreadLocal<StatusCommitFaultPoint> STATUS_COMMIT_FAULT = new ThreadLocal<>();
     private static final String SIGN_FINGERPRINT_PREFIX = "signfp:v1:";
     private static final int SIGN_FINGERPRINT_HEX_LENGTH = 64;
     private static final long INITIAL_FENCE = 0L;
@@ -250,7 +255,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
             FsOperations.ensureDir(this.paths.transactionalMetadataLog().getParent());
             openedMetadata = openMetadataStore();
             this.metadataStore = openedMetadata;
-            this.stagedContent.bindSigningOwnership(this::findSigningOwner);
+            this.stagedContent.bindTransactionalOwnership(this::findTransactionalOwners);
             this.credentialContentTransactions = new CredentialContentTransaction(this.paths, this.stagedContent);
             this.signingTimeWatermark = new AtomicLong(loadSigningTimeWatermark());
             this.historySeq = new AtomicLong(0L);
@@ -313,6 +318,17 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
         }
     }
 
+    private Set<DurableContentOwner> findTransactionalOwners(DurableContentReference reference) throws IOException {
+        Set<DurableContentOwner> owners = new HashSet<>(findSigningOwner(reference));
+        for (StoredStatus stored : listStoredStatuses()) {
+            if (reference.equals(stored.status().content())) {
+                owners.add(new DurableContentOwner(DurableContentOwner.Category.STATUS_OBJECT_RECORD,
+                        stored.status().statusObjectId().value()));
+            }
+        }
+        return Set.copyOf(owners);
+    }
+
     private Optional<DurableContentOwner> matchingSigningOwner(MetadataSnapshot.Record owner,
             DurableContentReference reference)
             throws IOException {
@@ -330,13 +346,15 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
 
     private void recoverStagedContent() throws IOException {
         credentialContentTransactions.recover();
+        List<StoredStatus> statuses = listStoredStatuses();
         try (TemporaryUniqueIndex retained = stagedContent.beginUniqueIndex();
-                TemporaryUniqueIndex retainedOwners = stagedContent.beginOwnerIndex()) {
+                TemporaryUniqueIndex retainedOwners = stagedContent.beginOwnerIndex();
+                TemporaryUniqueIndex transactionalOwners = stagedContent.beginUniqueIndex()) {
             try {
                 addPersistedCredentialReferences(retained, retainedOwners);
-                addPersistedStatusReferences(retained);
+                addTransactionalStatusOwnership(transactionalOwners, statuses);
                 addPendingSigningReferences(retained);
-                stagedContent.recoverContent(retained, retainedOwners);
+                stagedContent.recoverContent(retained, retainedOwners, transactionalOwners);
             } catch (IllegalStateException | PkiException malformedDurableState) {
                 // Recovery cannot prove abandonment while durable metadata is
                 // corrupt. Preserve content so the normal owning subsystem can
@@ -366,18 +384,11 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
         }
     }
 
-    private void addPersistedStatusReferences(TemporaryUniqueIndex retained) throws IOException {
+    private static void addTransactionalStatusOwnership(TemporaryUniqueIndex transactionalOwners,
-        Path root = paths.root().resolve("status").resolve("by-id");
+            List<StoredStatus> statuses)
-        if (!Files.isDirectory(root)) {
+            throws IOException {
-            return;
+        for (StoredStatus stored : statuses) {
-        }
+            addRetained(transactionalOwners, stored.status().content());
-        try (Stream<Path> pathsStream = Files.list(root)) {
-            java.util.Iterator<Path> iterator = pathsStream.filter(Files::isRegularFile).iterator();
-            while (iterator.hasNext()) {
-                StatusObject status = FsCodec.decode(FsCodec.STATUS_OBJECT, FsOperations.readAll(iterator.next()),
-                        stagedContent);
-                addRetained(retained, status.content());
-            }
         }
     }
 
@@ -430,15 +441,17 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
     /* default */ Set<String> snapshotNonCredentialContentIds() {
         requireStoreUsable();
         Set<String> contentIds = new HashSet<>();
-        for (StatusObject status : listBinaryFiles(paths.statusRoot(), FsCodec.STATUS_OBJECT)) {
-            contentIds.add(status.content().contentId());
-        }
         for (StoredSign stored : listStoredSigns()) {
             stored.reference().map(DurableContentReference::contentId).ifPresent(contentIds::add);
         }
         return Set.copyOf(contentIds);
     }
 
+    /* default */ List<StatusObject> snapshotStatusObjects() {
+        requireStoreUsable();
+        return listStoredStatuses().stream().map(StoredStatus::status).toList();
+    }
+
     @Override
     public void putCa(final CaRecord record) {
         requireStoreUsable();
@@ -596,16 +609,45 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
     public void putStatusObject(final StatusObject object) {
         requireStoreUsable();
         Objects.requireNonNull(object, "object");
-        PkiId id = object.statusObjectId();
+        byte[] encoded = validateStatusObject(object);
-        writeOnce(this.paths.statusObjectPath(id), FsCodec.encode(FsCodec.STATUS_OBJECT, object), "STATUS_OBJECT",
+        DurableContentReference reference = object.content();
-                FsUtil.safeId(id));
+        try (FilesystemStagedContentStore.TransactionalReservation reservation =
+                stagedContent.reserveTransactionalPublication(reference)) {
+            validateStatusContent(reference);
+            MetadataCommitResult result = createStatusMetadata(object, encoded);
+            if (result.outcome() == MetadataCommitResult.Outcome.COMMITTED) {
+                return;
+            }
+            if (result.outcome() == MetadataCommitResult.Outcome.UNKNOWN) {
+                reservation.preserveUntilRecovery();
+                durabilityUncertain.set(true);
+                throw new PkiException("Store durability unconfirmed: code=STORE_DURABILITY_UNCONFIRMED");
+            }
+            throw new IllegalStateException("Status object already exists");
+        } catch (IOException exception) {
+            throw new IllegalStateException("Status object content is invalid", exception);
+        }
     }
 
     @Override
     public Optional<StatusObject> getStatusObject(final PkiId statusObjectId) {
         requireStoreUsable();
         Objects.requireNonNull(statusObjectId, "statusObjectId");
-        return readOptional(this.paths.statusObjectPath(statusObjectId), FsCodec.STATUS_OBJECT);
+        try (MetadataSnapshot snapshot = metadataStore.snapshot()) {
+            Optional<MetadataSnapshot.Record> record = snapshot.get(statusRecordKey(statusObjectId));
+            if (record.isEmpty()) {
+                Optional<MetadataSnapshot.Record> owner = snapshot.get(statusOwnerKey(statusObjectId));
+                if (owner.isPresent()) {
+                    try (MetadataSnapshot.Record ignored = owner.orElseThrow()) {
+                        throw new IllegalStateException("Status owner exists without its record");
+                    }
+                }
+                return Optional.empty();
+            }
+            return Optional.of(decodeStoredStatus(snapshot, record.orElseThrow()).status());
+        } catch (IOException exception) {
+            throw new IllegalStateException("Failed to read authoritative status object", exception);
+        }
     }
 
     @Override
@@ -613,13 +655,9 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
         requireStoreUsable();
         Objects.requireNonNull(issuerCaId, "issuerCaId");
 
-        // Deterministic but coarse: scan all and filter by issuer id.
-        // This is acceptable for a reference implementation; indexes can be added
-        // later.
-        Path byId = this.paths.root().resolve("status").resolve("by-id");
-        List<StatusObject> all = listBinaryFiles(byId, FsCodec.STATUS_OBJECT);
         List<StatusObject> out = new ArrayList<>();
-        for (StatusObject o : all) {
+        for (StoredStatus stored : listStoredStatuses()) {
+            StatusObject o = stored.status();
             if (issuerCaId.equals(o.issuerCaId())) {
                 out.add(o);
             }
@@ -896,8 +934,8 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
                     stagedContent);
             DurableContentReference reference = continuation.content();
             MetadataCommitResult result;
-            try (FilesystemStagedContentStore.SigningReservation ignoredReservation =
+            try (FilesystemStagedContentStore.TransactionalReservation ignoredReservation =
-                            stagedContent.reserveSigningPublication(reference);
+                            stagedContent.reserveTransactionalPublication(reference);
                     RepeatableContent ignoredContent = stagedContent.openContent(reference)) {
                 stagedContent.restoreReference(reference.storeId(), reference.contentId(), reference.encoding(),
                         reference.length(), reference.sha256(), reference.lifecycle());
@@ -975,6 +1013,128 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
         }
     }
 
+    private List<StoredStatus> listStoredStatuses() {
+        List<StoredStatus> statuses = new ArrayList<>();
+        Set<String> recordIdentities = new HashSet<>();
+        try (MetadataSnapshot snapshot = metadataStore.snapshot()) {
+            try (MetadataCursor cursor = snapshot.scan(MetadataSnapshot.KeyRange.all(STATUS_RECORD_NAMESPACE),
+                    CancellationSignal.NONE)) {
+                Optional<MetadataSnapshot.Record> next;
+                while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) {
+                    StoredStatus stored = decodeStoredStatus(snapshot, next.orElseThrow());
+                    statuses.add(stored);
+                    recordIdentities.add(stored.status().statusObjectId().value());
+                }
+            }
+            try (MetadataCursor cursor = snapshot.scan(MetadataSnapshot.KeyRange.all(STATUS_OWNER_NAMESPACE),
+                    CancellationSignal.NONE)) {
+                Optional<MetadataSnapshot.Record> next;
+                while ((next = cursor.next(CancellationSignal.NONE)).isPresent()) {
+                    try (MetadataSnapshot.Record owner = next.orElseThrow()) {
+                        if (!recordIdentities.contains(owner.key().key())) {
+                            throw new IllegalStateException("Status owner exists without its record");
+                        }
+                    }
+                }
+            }
+            return List.copyOf(statuses);
+        } catch (IOException exception) {
+            throw new IllegalStateException("Failed to list authoritative status objects", exception);
+        }
+    }
+
+    private StoredStatus decodeStoredStatus(MetadataSnapshot snapshot, MetadataSnapshot.Record storedRecord)
+            throws IOException {
+        try (storedRecord) {
+            StatusObject status = FsCodec.decode(FsCodec.STATUS_OBJECT, readMetadataValue(storedRecord),
+                    stagedContent);
+            PkiId statusId = status.statusObjectId();
+            if (!statusRecordKey(statusId).equals(storedRecord.key())) {
+                throw new IOException("Status record metadata key mismatch");
+            }
+            validateStatusObject(status);
+            Optional<MetadataSnapshot.Record> storedOwner = snapshot.get(statusOwnerKey(statusId));
+            try (MetadataSnapshot.Record owner = storedOwner.orElseThrow(
+                    () -> new IOException("Status record owner edge is missing"))) {
+                DurableContentReference reference = decodeStatusOwner(owner, statusId);
+                if (!reference.equals(status.content())) {
+                    throw new IOException("Status record owner reference mismatch");
+                }
+                validateStatusContent(reference);
+                return new StoredStatus(status, storedRecord.recordRevision(), owner.recordRevision());
+            }
+        }
+    }
+
+    private byte[] validateStatusObject(StatusObject object) {
+        DurableContentReference reference = object.content();
+        if (!stagedContent.contentStoreId().equals(reference.storeId())
+                || reference.lifecycle() != DurableContentReference.Lifecycle.PERSISTED) {
+            throw new IllegalArgumentException("Status content reference is not persistent store authority");
+        }
+        byte[] encoded = FsCodec.encode(FsCodec.STATUS_OBJECT, object);
+        StatusObject canonical = FsCodec.decode(FsCodec.STATUS_OBJECT, encoded, stagedContent);
+        if (!object.statusObjectId().equals(canonical.statusObjectId())
+                || !Arrays.equals(encoded, FsCodec.encode(FsCodec.STATUS_OBJECT, canonical))) {
+            throw new IllegalArgumentException("Status object is not canonical");
+        }
+        return encoded;
+    }
+
+    private void validateStatusContent(DurableContentReference reference) throws IOException {
+        stagedContent.restoreReference(reference.storeId(), reference.contentId(), reference.encoding(),
+                reference.length(), reference.sha256(), reference.lifecycle());
+        try (RepeatableContent content = stagedContent.openContent(reference);
+                InputStream input = content.openStream()) {
+            byte[] buffer = new byte[METADATA_TRANSFER_BUFFER_BYTES];
+            while (true) {
+                int count = input.read(buffer);
+                if (count < 0) {
+                    return;
+                }
+                if (count == 0) {
+                    throw new IOException("Status content validation made no progress");
+                }
+            }
+        }
+    }
+
+    private MetadataCommitResult createStatusMetadata(StatusObject status, byte[] encoded) throws IOException {
+        PkiId statusId = status.statusObjectId();
+        try (MetadataTransaction transaction = metadataStore.beginTransaction()) {
+            transaction.create(statusRecordKey(statusId), byteContent(encoded), CancellationSignal.NONE);
+            transaction.create(statusOwnerKey(statusId),
+                    byteContent(encodeStatusOwner(statusId, status.content())), CancellationSignal.NONE);
+            if (STATUS_COMMIT_FAULT.get() == StatusCommitFaultPoint.BEFORE_COMMIT) {
+                throw new IOException("Injected status commit failure");
+            }
+            MetadataCommitResult result = transaction.commit();
+            if (result.outcome() == MetadataCommitResult.Outcome.COMMITTED
+                    && STATUS_COMMIT_FAULT.get() == StatusCommitFaultPoint.AFTER_COMMIT_AS_UNKNOWN) {
+                return new MetadataCommitResult(result.transactionId(), MetadataCommitResult.Outcome.UNKNOWN,
+                        OptionalLong.empty(), Optional.empty());
+            }
+            return result;
+        }
+    }
+
+    /* package */ static void installStatusCommitFault(StatusCommitFaultPoint point) {
+        STATUS_COMMIT_FAULT.set(Objects.requireNonNull(point, "point"));
+    }
+
+    /* package */ static void clearStatusCommitFault() {
+        STATUS_COMMIT_FAULT.remove();
+    }
+
+    /** Test-only status-publication boundary; never part of production API. */
+    /* default */
+    enum StatusCommitFaultPoint {
+        /** Fails after admission but before the metadata commit attempt. */
+        BEFORE_COMMIT,
+        /** Reports uncertainty only after a real durable commit. */
+        AFTER_COMMIT_AS_UNKNOWN
+    }
+
     @Override
     public Optional<SignWorkflowStore.Record> tryClaimSign(PkiId submissionId, long expectedRevision, Duration lease) {
         requireStoreUsable();
@@ -1693,6 +1853,14 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
         return new MetadataKey(SIGN_OWNER_NAMESPACE, parsed.id().value());
     }
 
+    private static MetadataKey statusRecordKey(PkiId statusObjectId) {
+        return new MetadataKey(STATUS_RECORD_NAMESPACE, statusObjectId.value());
+    }
+
+    private static MetadataKey statusOwnerKey(PkiId statusObjectId) {
+        return new MetadataKey(STATUS_OWNER_NAMESPACE, statusObjectId.value());
+    }
+
     private static RepeatableContent byteContent(byte[] value) {
         return new ByteValueContent(value);
     }
@@ -1700,7 +1868,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
     private static byte[] readMetadataValue(MetadataSnapshot.Record record) throws IOException {
         long length = record.length().orElseThrow();
         if (length < 0L || length > FsCodec.MAX_COMPONENT_BYTES) {
-            throw new IOException("Signing metadata value length is invalid");
+            throw new IOException("Metadata value length is invalid");
         }
         byte[] result = new byte[Math.toIntExact(length)];
         try (InputStream input = record.openStream()) {
@@ -1708,21 +1876,33 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
             while (offset < result.length) {
                 int count = input.read(result, offset, result.length - offset);
                 if (count <= 0) {
-                    throw new IOException("Signing metadata value is truncated");
+                    throw new IOException("Metadata value is truncated");
                 }
                 offset += count;
             }
             if (input.read() >= 0) {
-                throw new IOException("Signing metadata value has trailing data");
+                throw new IOException("Metadata value has trailing data");
             }
         }
         return result;
     }
 
     private static byte[] encodeSigningOwner(PkiId submissionId, DurableContentReference reference) {
+        return encodeTransactionalOwner(SIGN_OWNER_VALUE_VERSION, DurableContentOwner.Category.SIGNING_OPERATION,
+                submissionId, reference);
+    }
+
+    private static byte[] encodeStatusOwner(PkiId statusObjectId, DurableContentReference reference) {
+        new DurableContentOwner(DurableContentOwner.Category.STATUS_OBJECT_RECORD, statusObjectId.value());
+        return encodeTransactionalOwner(STATUS_OWNER_VALUE_VERSION,
+                DurableContentOwner.Category.STATUS_OBJECT_RECORD, statusObjectId, reference);
+    }
+
+    private static byte[] encodeTransactionalOwner(int version, DurableContentOwner.Category category,
+            PkiId ownerId, DurableContentReference reference) {
         byte[][] fields = {
-                DurableContentOwner.Category.SIGNING_OPERATION.name().getBytes(StandardCharsets.US_ASCII),
+                category.name().getBytes(StandardCharsets.US_ASCII),
-                submissionId.value().getBytes(StandardCharsets.UTF_8),
+                ownerId.value().getBytes(StandardCharsets.UTF_8),
                 reference.storeId().getBytes(StandardCharsets.UTF_8),
                 reference.contentId().getBytes(StandardCharsets.UTF_8),
                 reference.encoding().name().getBytes(StandardCharsets.US_ASCII),
@@ -1734,7 +1914,7 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
             size = Math.addExact(size, Math.addExact(Integer.BYTES, field.length));
         }
         ByteBuffer output = ByteBuffer.allocate(size);
-        output.putInt(SIGN_OWNER_VALUE_VERSION);
+        output.putInt(version);
         for (byte[] field : fields) {
             output.putInt(field.length).put(field);
         }
@@ -1744,10 +1924,23 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
 
     private DurableContentReference decodeSigningOwner(MetadataSnapshot.Record owner, PkiId expectedId)
             throws IOException {
+        return decodeTransactionalOwner(owner, expectedId, SIGN_OWNER_VALUE_VERSION,
+                DurableContentOwner.Category.SIGNING_OPERATION, signingOwnerKey(expectedId), "signing");
+    }
+
+    private DurableContentReference decodeStatusOwner(MetadataSnapshot.Record owner, PkiId expectedId)
+            throws IOException {
+        return decodeTransactionalOwner(owner, expectedId, STATUS_OWNER_VALUE_VERSION,
+                DurableContentOwner.Category.STATUS_OBJECT_RECORD, statusOwnerKey(expectedId), "status");
+    }
+
+    private DurableContentReference decodeTransactionalOwner(MetadataSnapshot.Record owner, PkiId expectedId,
+            int expectedVersion, DurableContentOwner.Category expectedCategory, MetadataKey expectedKey,
+            String authority) throws IOException {
         ByteBuffer input = ByteBuffer.wrap(readMetadataValue(owner));
         try {
-            if (input.getInt() != SIGN_OWNER_VALUE_VERSION) {
+            if (input.getInt() != expectedVersion) {
-                throw new IOException("Unsupported signing owner metadata");
+                throw new IOException("Unsupported " + authority + " owner metadata");
             }
             DurableContentOwner.Category category = DurableContentOwner.Category.valueOf(readOwnerField(input));
             String ownerId = readOwnerField(input);
@@ -1758,18 +1951,19 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
             DurableContentReference.Lifecycle lifecycle = DurableContentReference.Lifecycle.valueOf(
                     readOwnerField(input));
             long length = input.getLong();
-            if (input.hasRemaining() || category != DurableContentOwner.Category.SIGNING_OPERATION
+            if (input.hasRemaining() || category != expectedCategory
                     || !ownerId.equals(expectedId.value())) {
-                throw new IOException("Signing owner metadata identity mismatch");
+                throw new IOException("Transactional owner metadata identity mismatch");
             }
+            new DurableContentOwner(category, ownerId);
             DurableContentReference restored = stagedContent.restoreReference(storeId, contentId, encoding,
                     length, digest, lifecycle);
-            if (!signingOwnerKey(expectedId).equals(owner.key())) {
+            if (!expectedKey.equals(owner.key())) {
-                throw new IOException("Signing owner metadata key mismatch");
+                throw new IOException("Transactional owner metadata key mismatch");
             }
             return restored;
         } catch (IllegalArgumentException | java.nio.BufferUnderflowException exception) {
-            throw new IOException("Malformed signing owner metadata", exception);
+            throw new IOException("Malformed transactional owner metadata", exception);
         }
     }
 
@@ -1791,6 +1985,9 @@ public final class FilesystemPkiStore implements PkiStore, Closeable {
             OptionalLong ownerRevision, Optional<DurableContentReference> reference) {
     }
 
+    private record StoredStatus(StatusObject status, long recordRevision, long ownerRevision) {
+    }
+
     /** Store-owned bounded finite control metadata used only for synchronous admission. */
     private static final class ByteValueContent implements RepeatableContent {
         private final byte[] value;

pki/src/main/java/zeroecho/pki/impl/fs/FilesystemStagedContentStore.java

@@ -103,8 +103,8 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
     private final Path root;
     private final String storeId;
     private final ReentrantLock[] ownerLocks;
-    private final ConcurrentMap<String, Integer> signingReservations;
+    private final ConcurrentMap<String, Integer> transactionalReservations;
-    private final AtomicReference<SigningOwnership> signingOwnership;
+    private final AtomicReference<TransactionalOwnership> transactionalOwnership;
 
     /**
      * Creates a staged-content store.
@@ -118,8 +118,8 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         this.root = Objects.requireNonNull(root, "root").toAbsolutePath().normalize();
         this.storeId = requireStoreIdentifier(storeId);
         this.ownerLocks = new ReentrantLock[OWNER_LOCK_COUNT];
-        this.signingReservations = new ConcurrentHashMap<>();
+        this.transactionalReservations = new ConcurrentHashMap<>();
-        this.signingOwnership = new AtomicReference<>(reference -> Set.of());
+        this.transactionalOwnership = new AtomicReference<>(reference -> Set.of());
         for (int index = 0; index < ownerLocks.length; index++) {
             ownerLocks[index] = new ReentrantLock();
         }
@@ -184,7 +184,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
     public boolean retainContent(DurableContentReference reference, DurableContentOwner owner) throws IOException {
         DurableContentReference exact = requireOwned(reference);
         Objects.requireNonNull(owner, "owner");
-        SigningOwnershipIo.requireSidecarOwner(owner);
+        TransactionalOwnershipIo.requireSidecarOwner(owner);
         ReentrantLock lock = ownerLock(exact.contentId());
         lock.lock();
         try {
@@ -204,7 +204,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
     public boolean releaseContent(DurableContentReference reference, DurableContentOwner owner) throws IOException {
         DurableContentReference exact = requireOwned(reference);
         Objects.requireNonNull(owner, "owner");
-        SigningOwnershipIo.requireSidecarOwner(owner);
+        TransactionalOwnershipIo.requireSidecarOwner(owner);
         ReentrantLock lock = ownerLock(exact.contentId());
         lock.lock();
         try {
@@ -217,7 +217,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
                 return false;
             }
             writeOwners(exact.contentId(), owners);
-            SigningOwnershipIo.retireIfUnowned(this, exact, owners);
+            TransactionalOwnershipIo.retireIfUnowned(this, exact, owners);
             return true;
         } finally {
             lock.unlock();
@@ -232,7 +232,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         try {
             requireExactMetadata(exact);
             Set<DurableContentOwner> owners = readOwners(exact.contentId());
-            owners.addAll(signingOwnership.get().findOwners(exact));
+            owners.addAll(transactionalOwnership.get().findOwners(exact));
             return Set.copyOf(owners);
         } finally {
             lock.unlock();
@@ -246,7 +246,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         lock.lock();
         try {
             requireExactMetadata(exact);
-            SigningOwnershipIo.requireUnowned(this, exact);
+            TransactionalOwnershipIo.requireUnowned(this, exact);
             retireFiles(exact);
         } finally {
             lock.unlock();
@@ -255,7 +255,15 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
 
     @Override
     public void recoverContent(TemporaryUniqueIndex retained, TemporaryUniqueIndex retainedOwners) throws IOException {
-        StoreIo.recoverContent(this, retained, retainedOwners);
+        try (TemporaryUniqueIndex transactionalOwners = beginUniqueIndex()) {
+            StoreIo.recoverContent(this, retained, retainedOwners, transactionalOwners);
+        }
+    }
+
+    /** Recovery input separating transactional authority from general references and sidecars. */
+    /* package */ void recoverContent(TemporaryUniqueIndex retained, TemporaryUniqueIndex retainedOwners,
+            TemporaryUniqueIndex transactionalOwners) throws IOException {
+        StoreIo.recoverContent(this, retained, retainedOwners, transactionalOwners);
     }
 
     @Override
@@ -316,47 +324,52 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
     }
 
     private void writeOwners(String contentId, Set<DurableContentOwner> owners) throws IOException {
-        SigningOwnershipIo.requireSidecarOwners(owners);
+        TransactionalOwnershipIo.requireSidecarOwners(owners);
         StoreIo.writeOwners(this, contentId, owners);
     }
 
     /* package */ void retireSigningContent(DurableContentReference reference) throws IOException {
-        SigningOwnershipIo.retireSigningContent(this, reference);
+        TransactionalOwnershipIo.retireTransactionalContent(this, reference);
     }
 
-    /* package */ void bindSigningOwnership(SigningOwnership ownership) {
+    /* package */ void bindTransactionalOwnership(TransactionalOwnership ownership) {
-        signingOwnership.set(Objects.requireNonNull(ownership, "ownership"));
+        transactionalOwnership.set(Objects.requireNonNull(ownership, "ownership"));
     }
 
-    /* package */ SigningReservation reserveSigningPublication(DurableContentReference reference)
+    /* package */ TransactionalReservation reserveTransactionalPublication(DurableContentReference reference)
             throws IOException {
-        return SigningOwnershipIo.reserveSigningPublication(this, reference);
+        return TransactionalOwnershipIo.reserveTransactionalPublication(this, reference);
     }
 
-    /** Internal callback to the transactional signing-owner authority. */
+    /** Internal callback to signing and status transactional owner authority. */
     /* default */
     @FunctionalInterface
-    interface SigningOwnership {
+    interface TransactionalOwnership {
-        /** Finds every exact signing owner for a durable content reference. */
+        /** Finds every exact transactional owner for a durable content reference. */
         Set<DurableContentOwner> findOwners(DurableContentReference reference) throws IOException;
     }
 
     /** Short-lived reference-counted reservation spanning validation and commit. */
-    /* package */ final class SigningReservation implements AutoCloseable {
+    /* package */ final class TransactionalReservation implements AutoCloseable {
         private final String contentId;
         private final AtomicBoolean closed = new AtomicBoolean();
+        private final AtomicBoolean recoveryRequired = new AtomicBoolean();
 
-        private SigningReservation(String contentId) {
+        private TransactionalReservation(String contentId) {
             this.contentId = contentId;
         }
 
+        /* package */ void preserveUntilRecovery() {
+            recoveryRequired.set(true);
+        }
+
         @Override
         public void close() {
-            if (closed.compareAndSet(false, true)) {
+            if (!recoveryRequired.get() && closed.compareAndSet(false, true)) {
                 ReentrantLock lock = ownerLock(contentId);
                 lock.lock();
                 try {
-                    signingReservations.computeIfPresent(contentId,
+                    transactionalReservations.computeIfPresent(contentId,
                             (ignored, count) -> count == 1 ? null : count - 1);
                 } finally {
                     lock.unlock();
@@ -688,11 +701,13 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
     /** Mechanical home for filesystem metadata branches kept outside the store coordinator. */
     private static final class StoreIo {
         private static void recoverContent(FilesystemStagedContentStore store, TemporaryUniqueIndex retained,
-                TemporaryUniqueIndex retainedOwners) throws IOException {
+                TemporaryUniqueIndex retainedOwners, TemporaryUniqueIndex transactionalOwners) throws IOException {
             Objects.requireNonNull(retained, "retained");
             Objects.requireNonNull(retainedOwners, "retainedOwners");
+            Objects.requireNonNull(transactionalOwners, "transactionalOwners");
             retained.validateNamespace();
             retainedOwners.validateNamespace();
+            transactionalOwners.validateNamespace();
             try (java.util.stream.Stream<Path> paths = Files.list(store.root)) {
                 java.util.Iterator<Path> iterator = paths
                         .filter(path -> path.getFileName().toString().endsWith(".meta")).iterator();
@@ -700,7 +715,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
                     Path metadata = iterator.next();
                     String name = metadata.getFileName().toString();
                     String contentId = name.substring(0, name.length() - ".meta".length());
-                    recoverContent(store, retained, retainedOwners, contentId);
+                    recoverContent(store, retained, retainedOwners, transactionalOwners, contentId);
                 }
             }
             store.cleanupOrphanedCompletedFiles();
@@ -708,7 +723,8 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         }
 
         private static void recoverContent(FilesystemStagedContentStore store, TemporaryUniqueIndex retained,
-                TemporaryUniqueIndex retainedOwners, String contentId) throws IOException {
+                TemporaryUniqueIndex retainedOwners, TemporaryUniqueIndex transactionalOwners, String contentId)
+                throws IOException {
             ReentrantLock lock = store.ownerLock(contentId);
             lock.lock();
             try {
@@ -719,8 +735,10 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
                     store.writeOwners(contentId, owners);
                 }
                 boolean referenced = retained.contains(contentId.getBytes(StandardCharsets.US_ASCII));
+                boolean transactionallyOwned = transactionalOwners.contains(
+                        contentId.getBytes(StandardCharsets.US_ASCII));
                 boolean keep = reference.lifecycle() != DurableContentReference.Lifecycle.TEMPORARY
-                        && (referenced || !owners.isEmpty());
+                        && (referenced || transactionallyOwned || !owners.isEmpty());
                 if (!keep || !isRegularFile(store.completePath(reference))) {
                     store.retireFiles(reference);
                 }
@@ -863,7 +881,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
             }
             for (int index = 0; index < count; index++) {
                 DurableContentOwner owner = parseOwner(input.readUTF());
-                if (owner.category() == DurableContentOwner.Category.SIGNING_OPERATION) {
+                if (TransactionalOwnershipIo.isTransactionalOwner(owner)) {
                     continue;
                 }
                 if (!owners.add(owner)) {
@@ -933,8 +951,8 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         }
     }
 
-    /** Isolates transactional signing-ownership lifecycle decisions. */
+    /** Isolates transactional staged-content ownership lifecycle decisions. */
-    private static final class SigningOwnershipIo {
+    private static final class TransactionalOwnershipIo {
         private static void requireSidecarOwners(Set<DurableContentOwner> owners) {
             for (DurableContentOwner owner : owners) {
                 requireSidecarOwner(owner);
@@ -942,37 +960,42 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         }
 
         private static void requireSidecarOwner(DurableContentOwner owner) {
-            if (owner.category() == DurableContentOwner.Category.SIGNING_OPERATION) {
+            if (isTransactionalOwner(owner)) {
                 throw new IllegalArgumentException(
-                        "Signing-operation ownership is authoritative transactional metadata");
+                        "Transactional ownership is authoritative metadata");
             }
         }
 
-        private static SigningReservation reserveSigningPublication(FilesystemStagedContentStore store,
+        private static boolean isTransactionalOwner(DurableContentOwner owner) {
+            return owner.category() == DurableContentOwner.Category.SIGNING_OPERATION
+                    || owner.category() == DurableContentOwner.Category.STATUS_OBJECT_RECORD;
+        }
+
+        private static TransactionalReservation reserveTransactionalPublication(FilesystemStagedContentStore store,
                 DurableContentReference reference) throws IOException {
             DurableContentReference exact = store.requireOwned(reference);
             ReentrantLock lock = store.ownerLock(exact.contentId());
             lock.lock();
             try {
                 store.requireExactMetadata(exact);
-                store.signingReservations.merge(exact.contentId(), 1, Math::addExact);
+                store.transactionalReservations.merge(exact.contentId(), 1, Math::addExact);
-                return store.new SigningReservation(exact.contentId());
+                return store.new TransactionalReservation(exact.contentId());
             } catch (ArithmeticException exception) {
-                throw new IOException("Signing content reservation limit exceeded", exception);
+                throw new IOException("Transactional content reservation limit exceeded", exception);
             } finally {
                 lock.unlock();
             }
         }
 
-        private static boolean hasSigningAuthority(FilesystemStagedContentStore store,
+        private static boolean hasTransactionalAuthority(FilesystemStagedContentStore store,
                 DurableContentReference reference) throws IOException {
-            return store.signingReservations.containsKey(reference.contentId())
+            return store.transactionalReservations.containsKey(reference.contentId())
-                    || !store.signingOwnership.get().findOwners(reference).isEmpty();
+                    || !store.transactionalOwnership.get().findOwners(reference).isEmpty();
         }
 
         private static void retireIfUnowned(FilesystemStagedContentStore store,
                 DurableContentReference reference, Set<DurableContentOwner> owners) throws IOException {
-            if (owners.isEmpty() && !hasSigningAuthority(store, reference)) {
+            if (owners.isEmpty() && !hasTransactionalAuthority(store, reference)) {
                 store.retireFiles(reference);
             }
         }
@@ -980,12 +1003,12 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
         private static void requireUnowned(FilesystemStagedContentStore store,
                 DurableContentReference reference) throws IOException {
             if (!store.readOwners(reference.contentId()).isEmpty()
-                    || hasSigningAuthority(store, reference)) {
+                    || hasTransactionalAuthority(store, reference)) {
                 throw new IOException("Staged content remains durably owned");
             }
         }
 
-        private static void retireSigningContent(FilesystemStagedContentStore store,
+        private static void retireTransactionalContent(FilesystemStagedContentStore store,
                 DurableContentReference reference) throws IOException {
             DurableContentReference exact = store.requireOwned(reference);
             ReentrantLock lock = store.ownerLock(exact.contentId());
@@ -993,7 +1016,7 @@ public final class FilesystemStagedContentStore implements StagedContentStore {
             try {
                 store.requireExactMetadata(exact);
                 if (store.readOwners(exact.contentId()).isEmpty()
-                        && !hasSigningAuthority(store, exact)) {
+                        && !hasTransactionalAuthority(store, exact)) {
                     store.retireFiles(exact);
                 }
             } finally {

pki/src/main/java/zeroecho/pki/impl/fs/FsPaths.java

@@ -177,19 +177,6 @@ final class FsPaths {
         return this.root.resolve("revocation-snapshots");
     }
 
-    // -------------------------------------------------------------------------
-    // Status objects (immutable .bin)
-    // -------------------------------------------------------------------------
-
-    /* default */ Path statusObjectPath(final PkiId statusObjectId) {
-        Objects.requireNonNull(statusObjectId, "statusObjectId");
-        return this.root.resolve("status").resolve(BY_ID).resolve(FsUtil.safeId(statusObjectId) + ".bin");
-    }
-
-    /* default */ Path statusRoot() {
-        return this.root.resolve("status").resolve(BY_ID);
-    }
-
     // -------------------------------------------------------------------------
     // Policy traces (immutable .bin)
     // -------------------------------------------------------------------------

pki/src/main/java/zeroecho/pki/impl/fs/FsSnapshotExporter.java

@@ -60,6 +60,7 @@ import zeroecho.pki.api.PkiId;
 import zeroecho.pki.api.ca.CaRecord;
 import zeroecho.pki.api.content.DurableContentReference;
 import zeroecho.pki.api.credential.Credential;
+import zeroecho.pki.api.status.StatusObject;
 import zeroecho.pki.impl.ProfileLifecycleFailure;
 import zeroecho.pki.impl.ProfileLifecycleFailure.Code;
 import zeroecho.pki.spi.store.ContentSink;
@@ -134,7 +135,7 @@ final class FsSnapshotExporter {
         }
     }
 
-    private static DurableContentReference transferCredentialContent(FilesystemPkiStore source,
+    private static DurableContentReference transferContent(FilesystemPkiStore source,
             FilesystemPkiStore target, DurableContentReference sourceReference) throws IOException {
         try (RepeatableContent content = source.stagedContent().openContent(sourceReference);
                 InputStream input = content.openStream();
@@ -212,7 +213,12 @@ final class FsSnapshotExporter {
         private SnapshotAuthority plan(Instant at) throws IOException {
             CredentialInventory inventory = inventoryCredentials();
             List<CaRecord> cas = selectCas(at, inventory.credentials());
-            return new SnapshotAuthority(cas, inventory.credentials(), inventory.contentIds());
+            List<StatusObject> statuses = source.snapshotStatusObjects();
+            Set<String> remintedContentIds = new HashSet<>(inventory.contentIds());
+            for (StatusObject status : statuses) {
+                remintedContentIds.add(status.content().contentId());
+            }
+            return new SnapshotAuthority(cas, inventory.credentials(), statuses, remintedContentIds);
         }
 
         private CredentialInventory inventoryCredentials() throws IOException {
@@ -336,12 +342,11 @@ final class FsSnapshotExporter {
             copyFile(sourceRoot.resolve("SIGNING_NAMESPACE"), targetRoot.resolve("SIGNING_NAMESPACE"));
             copyFile(sourceRoot.resolve("SIGNING_TIME_WATERMARK"), targetRoot.resolve("SIGNING_TIME_WATERMARK"));
             copyTreeIfExists(sourceRoot.resolve("requests"), targetRoot.resolve("requests"));
-            copyTreeIfExists(sourceRoot.resolve("status"), targetRoot.resolve("status"));
             copyTreeIfExists(sourceRoot.resolve("policy"), targetRoot.resolve("policy"));
             copyTreeIfExists(sourceRoot.resolve("publications"), targetRoot.resolve("publications"));
             copyTreeIfExists(sourceRoot.resolve("sign-workflows"), targetRoot.resolve("sign-workflows"));
             copyStagedContent(sourceRoot.resolve("staged-content"), targetRoot.resolve("staged-content"),
-                    plan.authority().sourceCredentialContentIds(), plan.nonCredentialContentIds());
+                    plan.authority().remintedContentIds(), plan.nonCredentialContentIds());
             copyTreeIfExists(sourceRoot.resolve("revocations"), targetRoot.resolve("revocations"));
             copyImportedProfilesAsOf(plan.profiles(), targetRoot.resolve("profiles"));
             reconstructMutableTree(sourceRoot.resolve("workflows"), targetRoot.resolve("workflows"), at,
@@ -384,6 +389,9 @@ final class FsSnapshotExporter {
                         persistCa(target, ca);
                     }
                 }
+                for (StatusObject status : authority.statuses()) {
+                    persistStatus(target, status);
+                }
             }
         }
 
@@ -420,6 +428,12 @@ final class FsSnapshotExporter {
                 return false;
             }
         }
+
+        private void persistStatus(FilesystemPkiStore target, StatusObject sourceStatus) throws IOException {
+            try (StatusTransfer transfer = new StatusTransfer(source, target, sourceStatus)) {
+                transfer.persist();
+            }
+        }
     }
 
     /** Owns a completed target reference until its credential is persisted. */
@@ -433,7 +447,7 @@ final class FsSnapshotExporter {
                 Credential sourceCredential) throws IOException {
             this.target = target;
             this.sourceCredential = sourceCredential;
-            this.targetReference = transferCredentialContent(source, target, sourceCredential.content());
+            this.targetReference = transferContent(source, target, sourceCredential.content());
         }
 
         private void persist() {
@@ -449,6 +463,35 @@ final class FsSnapshotExporter {
         }
     }
 
+    /** Owns reminted target content until its status authority is committed. */
+    private static final class StatusTransfer implements AutoCloseable {
+        private final FilesystemPkiStore target;
+        private final StatusObject sourceStatus;
+        private final DurableContentReference targetReference;
+        private boolean persisted;
+
+        private StatusTransfer(FilesystemPkiStore source, FilesystemPkiStore target, StatusObject sourceStatus)
+                throws IOException {
+            this.target = target;
+            this.sourceStatus = sourceStatus;
+            this.targetReference = transferContent(source, target, sourceStatus.content());
+        }
+
+        private void persist() {
+            target.putStatusObject(new StatusObject(sourceStatus.statusObjectId(), sourceStatus.formatId(),
+                    sourceStatus.issuerCaId(), sourceStatus.type(), sourceStatus.thisUpdate(),
+                    sourceStatus.nextUpdate(), targetReference, sourceStatus.attributes()));
+            persisted = true;
+        }
+
+        @Override
+        public void close() throws IOException {
+            if (!persisted) {
+                target.stagedContent().retireUnownedContent(targetReference);
+            }
+        }
+    }
+
     private static void deleteOwnedTree(Path root) throws IOException {
         if (!Files.exists(root, LinkOption.NOFOLLOW_LINKS)) {
             return;
@@ -683,11 +726,12 @@ final class FsSnapshotExporter {
     }
 
     private record SnapshotAuthority(List<CaRecord> cas, Map<PkiId, Credential> credentials,
-            Set<String> sourceCredentialContentIds) {
+            List<StatusObject> statuses, Set<String> remintedContentIds) {
         private SnapshotAuthority {
             cas = List.copyOf(cas);
             credentials = Collections.unmodifiableMap(new LinkedHashMap<>(credentials));
-            sourceCredentialContentIds = Set.copyOf(sourceCredentialContentIds);
+            statuses = List.copyOf(statuses);
+            remintedContentIds = Set.copyOf(remintedContentIds);
         }
     }
 

pki/src/test/java/zeroecho/pki/impl/fs/FilesystemPkiStoreTest.java

@@ -40,7 +40,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.ByteArrayInputStream;
+import java.io.DataOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.MessageDigest;
@@ -49,6 +54,7 @@ import java.time.Instant;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
@@ -56,12 +62,16 @@ import java.util.stream.Collectors;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
+import zeroecho.core.io.CancellationSignal;
+import zeroecho.core.io.RepeatableContent;
 import zeroecho.pki.api.EncodedObject;
 import zeroecho.pki.api.content.DurableContentReference;
+import zeroecho.pki.api.content.DurableContentOwner;
 import zeroecho.pki.api.Encoding;
 import zeroecho.pki.api.FormatId;
 import zeroecho.pki.api.IssuerRef;
 import zeroecho.pki.api.KeyRef;
+import zeroecho.pki.api.PkiException;
 import zeroecho.pki.api.PkiId;
 import zeroecho.pki.api.SubjectRef;
 import zeroecho.pki.api.Validity;
@@ -102,6 +112,10 @@ import zeroecho.pki.api.revocation.RevocationJournal;
 import zeroecho.pki.api.revocation.RevocationReason;
 import zeroecho.pki.api.status.StatusObject;
 import zeroecho.pki.api.status.StatusObjectType;
+import zeroecho.pki.spi.store.MetadataCommitResult;
+import zeroecho.pki.spi.store.MetadataKey;
+import zeroecho.pki.spi.store.MetadataSnapshot;
+import zeroecho.pki.spi.store.MetadataTransaction;
 import zeroecho.pki.spi.store.PkiStore;
 
 /**
@@ -186,6 +200,317 @@ public final class FilesystemPkiStoreTest {
         System.out.println("...ok");
     }
 
+    @Test
+    void statusAuthorityIsAtomicDurableWriteOnceAndProtectsContent() throws Exception {
+        System.out.println("statusAuthorityIsAtomicDurableWriteOnceAndProtectsContent");
+        Path root = tmp.resolve("store-status-authority");
+        PkiId firstId = new PkiId("crl:status-authority-one");
+        PkiId secondId = new PkiId("crl:status-authority-two");
+        DurableContentReference firstReference;
+        StatusObject firstStatus;
+        try (FilesystemPkiStore store = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults())) {
+            firstReference = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(store.stagedContent(), Encoding.DER,
+                    new byte[] { 1, 2, 3 });
+            firstStatus = new StatusObject(firstId, new FormatId("fmt-x509"), new PkiId("ca-status"),
+                    StatusObjectType.CRL, Instant.EPOCH, Optional.empty(), firstReference,
+                    TestObjects.emptyAttributes());
+            store.putStatusObject(firstStatus);
+            assertThrows(IllegalStateException.class, () -> store.putStatusObject(firstStatus));
+            assertThrows(IOException.class, () -> store.stagedContent().retireUnownedContent(firstReference));
+
+            DurableContentReference conflictingReference = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(
+                    store.stagedContent(), Encoding.DER, new byte[] { 7, 8, 9 });
+            StatusObject conflicting = new StatusObject(firstId, new FormatId("fmt-x509"),
+                    new PkiId("ca-status"), StatusObjectType.CRL, Instant.EPOCH.plusSeconds(2L), Optional.empty(),
+                    conflictingReference, TestObjects.emptyAttributes());
+            assertThrows(IllegalStateException.class, () -> store.putStatusObject(conflicting));
+            assertEquals(firstReference, store.getStatusObject(firstId).orElseThrow().content());
+            store.stagedContent().retireUnownedContent(conflictingReference);
+
+            DurableContentReference secondReference = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(
+                    store.stagedContent(), Encoding.DER, new byte[] { 4, 5, 6 });
+            store.putStatusObject(new StatusObject(secondId, new FormatId("fmt-x509"), new PkiId("ca-status"),
+                    StatusObjectType.CRL, Instant.EPOCH.plusSeconds(1L), Optional.empty(), secondReference,
+                    TestObjects.emptyAttributes()));
+            assertEquals(2, store.listStatusObjects(new PkiId("ca-status")).size());
+            assertTrue(store.listPublicationRecords().isEmpty());
+            assertFalse(Files.exists(root.resolve("staged-content").resolve(firstReference.contentId() + ".owners")));
+        }
+
+        PkiId legacyId = new PkiId("crl:legacy-distinct");
+        StatusObject legacy = new StatusObject(legacyId, firstStatus.formatId(), firstStatus.issuerCaId(),
+                firstStatus.type(), firstStatus.thisUpdate(), firstStatus.nextUpdate(), firstReference,
+                firstStatus.attributes());
+        Path obsolete = root.resolve("status").resolve("by-id").resolve(FsUtil.safeId(legacyId) + ".bin");
+        Files.createDirectories(obsolete.getParent());
+        Files.write(obsolete, FsCodec.encode(FsCodec.STATUS_OBJECT, legacy));
+        try (FilesystemPkiStore reopened = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults())) {
+            assertEquals(firstId, reopened.getStatusObject(firstId).orElseThrow().statusObjectId());
+            assertEquals(secondId, reopened.getStatusObject(secondId).orElseThrow().statusObjectId());
+            assertTrue(reopened.getStatusObject(legacyId).isEmpty());
+            assertEquals(2, reopened.listStatusObjects(new PkiId("ca-status")).size());
+            assertThrows(IOException.class, () -> reopened.stagedContent().retireUnownedContent(firstReference));
+        }
+        System.out.println("...statusRecords=2");
+        System.out.println("statusAuthorityIsAtomicDurableWriteOnceAndProtectsContent...ok");
+    }
+
+    @Test
+    void statusRecoveryRejectsRecordOwnerAsymmetry() throws Exception {
+        System.out.println("statusRecoveryRejectsRecordOwnerAsymmetry");
+        Path missingOwnerRoot = tmp.resolve("store-status-missing-owner");
+        PkiId missingOwnerId = persistFixtureStatus(missingOwnerRoot, "missing-owner").statusObjectId();
+        deleteMetadataRecord(missingOwnerRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-owner", missingOwnerId.value()));
+        assertThrows(IllegalStateException.class,
+                () -> new FilesystemPkiStore(missingOwnerRoot, FsPkiStoreOptions.defaults()));
+
+        Path orphanOwnerRoot = tmp.resolve("store-status-orphan-owner");
+        PkiId orphanOwnerId = persistFixtureStatus(orphanOwnerRoot, "orphan-owner").statusObjectId();
+        deleteMetadataRecord(orphanOwnerRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-record", orphanOwnerId.value()));
+        assertThrows(IllegalStateException.class,
+                () -> new FilesystemPkiStore(orphanOwnerRoot, FsPkiStoreOptions.defaults()));
+        System.out.println("...rejected=missing-owner,orphan-owner");
+        System.out.println("statusRecoveryRejectsRecordOwnerAsymmetry...ok");
+    }
+
+    @Test
+    void statusRecoveryRejectsMalformedOrMismatchedMetadata() throws Exception {
+        System.out.println("statusRecoveryRejectsMalformedOrMismatchedMetadata");
+        Path wrongIdentityRoot = tmp.resolve("store-status-wrong-owner-id");
+        StatusObject wrongIdentity = persistFixtureStatus(wrongIdentityRoot, "wrong-owner-id");
+        replaceMetadataRecord(wrongIdentityRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-owner", wrongIdentity.statusObjectId().value()),
+                encodeStatusOwner("crl:different", wrongIdentity.content()));
+        assertStoreOpenFails(wrongIdentityRoot);
+
+        Path wrongReferenceRoot = tmp.resolve("store-status-wrong-owner-reference");
+        StatusObject wrongReference;
+        DurableContentReference alternate;
+        try (FilesystemPkiStore store = new FilesystemPkiStore(wrongReferenceRoot, FsPkiStoreOptions.defaults())) {
+            DurableContentReference original = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(
+                    store.stagedContent(), Encoding.DER, new byte[] { 1 });
+            wrongReference = new StatusObject(new PkiId("crl:wrong-owner-reference"), new FormatId("fmt-x509"),
+                    new PkiId("ca-status"), StatusObjectType.CRL, Instant.EPOCH, Optional.empty(), original,
+                    TestObjects.emptyAttributes());
+            store.putStatusObject(wrongReference);
+            alternate = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(store.stagedContent(), Encoding.DER,
+                    new byte[] { 2 });
+        }
+        replaceMetadataRecord(wrongReferenceRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-owner", wrongReference.statusObjectId().value()),
+                encodeStatusOwner(wrongReference.statusObjectId().value(), alternate));
+        assertStoreOpenFails(wrongReferenceRoot);
+
+        Path malformedRecordRoot = tmp.resolve("store-status-malformed-record");
+        StatusObject malformedRecord = persistFixtureStatus(malformedRecordRoot, "malformed-record");
+        replaceMetadataRecord(malformedRecordRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-record", malformedRecord.statusObjectId().value()),
+                new byte[] { 1, 2, 3 });
+        assertStoreOpenFails(malformedRecordRoot);
+
+        Path malformedOwnerRoot = tmp.resolve("store-status-malformed-owner");
+        StatusObject malformedOwner = persistFixtureStatus(malformedOwnerRoot, "malformed-owner");
+        replaceMetadataRecord(malformedOwnerRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-owner", malformedOwner.statusObjectId().value()),
+                new byte[] { 4, 5, 6 });
+        assertStoreOpenFails(malformedOwnerRoot);
+        System.out.println("...rejected=identity,reference,record,owner");
+        System.out.println("statusRecoveryRejectsMalformedOrMismatchedMetadata...ok");
+    }
+
+    @Test
+    void statusRecoveryRejectsMissingOrCorruptContentAndSidecarRepair() throws Exception {
+        System.out.println("statusRecoveryRejectsMissingOrCorruptContentAndSidecarRepair");
+        Path missingMetadataRoot = tmp.resolve("store-status-missing-meta");
+        StatusObject missingMetadata = persistFixtureStatus(missingMetadataRoot, "missing-meta");
+        Files.delete(missingMetadataRoot.resolve("staged-content")
+                .resolve(missingMetadata.content().contentId() + ".meta"));
+        assertStoreOpenFails(missingMetadataRoot);
+
+        Path missingContentRoot = tmp.resolve("store-status-missing-content");
+        StatusObject missingContent = persistFixtureStatus(missingContentRoot, "missing-content");
+        Files.delete(missingContentRoot.resolve("staged-content")
+                .resolve(missingContent.content().contentId() + ".content"));
+        assertStoreOpenFails(missingContentRoot);
+
+        Path corruptContentRoot = tmp.resolve("store-status-corrupt-content");
+        StatusObject corruptContent = persistFixtureStatus(corruptContentRoot, "corrupt-content");
+        Files.write(corruptContentRoot.resolve("staged-content")
+                .resolve(corruptContent.content().contentId() + ".content"),
+                new byte[Math.toIntExact(corruptContent.content().length())]);
+        assertStoreOpenFails(corruptContentRoot);
+
+        Path sidecarRoot = tmp.resolve("store-status-sidecar-repair");
+        StatusObject sidecar = persistFixtureStatus(sidecarRoot, "sidecar-repair");
+        deleteMetadataRecord(sidecarRoot,
+                new MetadataKey("io.zeroecho.pki.status-object-owner", sidecar.statusObjectId().value()));
+        writeStatusSidecar(sidecarRoot, sidecar);
+        assertStoreOpenFails(sidecarRoot);
+        System.out.println("...rejected=missing-meta,missing-content,corrupt-content,sidecar-repair");
+        System.out.println("statusRecoveryRejectsMissingOrCorruptContentAndSidecarRepair...ok");
+    }
+
+    @Test
+    void statusCommitFaultsPreserveAtomicityAndUnknownRecovery() throws Exception {
+        System.out.println("statusCommitFaultsPreserveAtomicityAndUnknownRecovery");
+        Path beforeCommitRoot = tmp.resolve("store-status-before-commit-failure");
+        try (FilesystemPkiStore store = new FilesystemPkiStore(beforeCommitRoot, FsPkiStoreOptions.defaults())) {
+            DurableContentReference reference = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(
+                    store.stagedContent(), Encoding.DER, new byte[] { 3, 4 });
+            StatusObject status = new StatusObject(new PkiId("crl:before-commit"), new FormatId("fmt-x509"),
+                    new PkiId("ca-status"), StatusObjectType.CRL, Instant.EPOCH, Optional.empty(), reference,
+                    TestObjects.emptyAttributes());
+            FilesystemPkiStore.installStatusCommitFault(
+                    FilesystemPkiStore.StatusCommitFaultPoint.BEFORE_COMMIT);
+            try {
+                assertThrows(IllegalStateException.class, () -> store.putStatusObject(status));
+            } finally {
+                FilesystemPkiStore.clearStatusCommitFault();
+            }
+            assertTrue(store.getStatusObject(status.statusObjectId()).isEmpty());
+            store.stagedContent().retireUnownedContent(reference);
+            assertThrows(IOException.class, () -> store.stagedContent().openContent(reference));
+        }
+
+        Path unknownRoot = tmp.resolve("store-status-after-commit-unknown");
+        PkiId unknownId = new PkiId("crl:after-commit-unknown");
+        DurableContentReference committedReference;
+        try (FilesystemPkiStore store = new FilesystemPkiStore(unknownRoot, FsPkiStoreOptions.defaults())) {
+            committedReference = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(store.stagedContent(),
+                    Encoding.DER, new byte[] { 5, 6 });
+            StatusObject status = new StatusObject(unknownId, new FormatId("fmt-x509"),
+                    new PkiId("ca-status"), StatusObjectType.CRL, Instant.EPOCH, Optional.empty(),
+                    committedReference, TestObjects.emptyAttributes());
+            FilesystemPkiStore.installStatusCommitFault(
+                    FilesystemPkiStore.StatusCommitFaultPoint.AFTER_COMMIT_AS_UNKNOWN);
+            try {
+                PkiException unknown = assertThrows(PkiException.class, () -> store.putStatusObject(status));
+                assertTrue(unknown.getMessage().contains("STORE_DURABILITY_UNCONFIRMED"));
+            } finally {
+                FilesystemPkiStore.clearStatusCommitFault();
+            }
+            assertThrows(PkiException.class, () -> store.getStatusObject(unknownId));
+            assertThrows(IOException.class, () -> store.stagedContent().retireUnownedContent(committedReference));
+        } finally {
+            FilesystemPkiStore.clearStatusCommitFault();
+        }
+        try (FilesystemPkiStore reopened = new FilesystemPkiStore(unknownRoot, FsPkiStoreOptions.defaults())) {
+            assertEquals(committedReference,
+                    reopened.getStatusObject(unknownId).orElseThrow().content());
+        }
+        System.out.println("...faults=before-commit,after-commit-unknown");
+        System.out.println("statusCommitFaultsPreserveAtomicityAndUnknownRecovery...ok");
+    }
+
+    private StatusObject persistFixtureStatus(Path root, String suffix) throws Exception {
+        PkiId statusId = new PkiId("crl:" + suffix);
+        try (FilesystemPkiStore store = new FilesystemPkiStore(root, FsPkiStoreOptions.defaults())) {
+            DurableContentReference reference = zeroecho.pki.testkit.PkiTestRuntime.fixtureReference(
+                    store.stagedContent(), Encoding.DER, suffix.getBytes(StandardCharsets.US_ASCII));
+            StatusObject status = new StatusObject(statusId, new FormatId("fmt-x509"), new PkiId("ca-status"),
+                    StatusObjectType.CRL, Instant.EPOCH, Optional.empty(), reference, TestObjects.emptyAttributes());
+            store.putStatusObject(status);
+            return status;
+        }
+    }
+
+    private static void deleteMetadataRecord(Path root, MetadataKey key) throws Exception {
+        FsPaths paths = new FsPaths(root);
+        try (PosixTransactionalMetadataStore metadata = PosixTransactionalMetadataStore.open(
+                paths.transactionalMetadataLog())) {
+            long revision;
+            try (MetadataSnapshot snapshot = metadata.snapshot();
+                    MetadataSnapshot.Record record = snapshot.get(key).orElseThrow()) {
+                revision = record.recordRevision();
+            }
+            try (MetadataTransaction transaction = metadata.beginTransaction()) {
+                transaction.delete(key, revision);
+                assertEquals(MetadataCommitResult.Outcome.COMMITTED, transaction.commit().outcome());
+            }
+        }
+    }
+
+    private static void replaceMetadataRecord(Path root, MetadataKey key, byte[] value) throws Exception {
+        FsPaths paths = new FsPaths(root);
+        try (PosixTransactionalMetadataStore metadata = PosixTransactionalMetadataStore.open(
+                paths.transactionalMetadataLog())) {
+            long revision;
+            try (MetadataSnapshot snapshot = metadata.snapshot();
+                    MetadataSnapshot.Record record = snapshot.get(key).orElseThrow()) {
+                revision = record.recordRevision();
+            }
+            try (MetadataTransaction transaction = metadata.beginTransaction()) {
+                transaction.replace(key, revision, byteContent(value), CancellationSignal.NONE);
+                assertEquals(MetadataCommitResult.Outcome.COMMITTED, transaction.commit().outcome());
+            }
+        }
+    }
+
+    private static RepeatableContent byteContent(byte[] value) {
+        byte[] immutable = value.clone();
+        return new RepeatableContent() {
+            @Override
+            public InputStream openStream() {
+                return new ByteArrayInputStream(immutable);
+            }
+
+            @Override
+            public OptionalLong length() {
+                return OptionalLong.of(immutable.length);
+            }
+
+            @Override
+            public String contentId() {
+                return "test-status-metadata";
+            }
+
+            @Override
+            public void close() {
+                // Immutable in-memory test content owns no external resource.
+            }
+        };
+    }
+
+    private static byte[] encodeStatusOwner(String ownerId, DurableContentReference reference) {
+        byte[][] fields = {
+                DurableContentOwner.Category.STATUS_OBJECT_RECORD.name().getBytes(StandardCharsets.US_ASCII),
+                ownerId.getBytes(StandardCharsets.UTF_8), reference.storeId().getBytes(StandardCharsets.UTF_8),
+                reference.contentId().getBytes(StandardCharsets.UTF_8),
+                reference.encoding().name().getBytes(StandardCharsets.US_ASCII),
+                reference.sha256().getBytes(StandardCharsets.US_ASCII),
+                reference.lifecycle().name().getBytes(StandardCharsets.US_ASCII)
+        };
+        int size = Integer.BYTES + Long.BYTES;
+        for (byte[] field : fields) {
+            size = Math.addExact(size, Math.addExact(Integer.BYTES, field.length));
+        }
+        ByteBuffer output = ByteBuffer.allocate(size);
+        output.putInt(1);
+        for (byte[] field : fields) {
+            output.putInt(field.length).put(field);
+        }
+        output.putLong(reference.length());
+        return output.array();
+    }
+
+    private static void writeStatusSidecar(Path root, StatusObject status) throws IOException {
+        Path owners = root.resolve("staged-content").resolve(status.content().contentId() + ".owners");
+        DurableContentOwner owner = new DurableContentOwner(DurableContentOwner.Category.STATUS_OBJECT_RECORD,
+                status.statusObjectId().value());
+        try (DataOutputStream output = new DataOutputStream(Files.newOutputStream(owners))) {
+            output.writeByte(2);
+            output.writeUTF(status.content().contentId());
+            output.writeInt(1);
+            output.writeUTF(owner.canonicalForm());
+        }
+    }
+
+    private static void assertStoreOpenFails(Path root) {
+        assertThrows(IllegalStateException.class, () -> new FilesystemPkiStore(root, FsPkiStoreOptions.defaults()));
+    }
+
     @Test
     void writeOnceCredentialRejected() throws Exception {
         System.out.println("writeOnceCredentialRejected");
@@ -365,12 +690,13 @@ public final class FilesystemPkiStoreTest {
             Credential credential = restored.getCredential(credentialId).orElseThrow();
             StatusObject status = restored.getStatusObject(statusId).orElseThrow();
             assertFalse(sourceContentId.equals(credential.content().contentId()));
-            assertEquals(sourceContentId, status.content().contentId());
+            assertFalse(sourceContentId.equals(status.content().contentId()));
+            assertFalse(credential.content().contentId().equals(status.content().contentId()));
             assertArrayEquals(expected, zeroecho.pki.testkit.PkiTestRuntime.readContent(restored, credential.content()));
             assertArrayEquals(expected, zeroecho.pki.testkit.PkiTestRuntime.readContent(restored, status.content()));
         }
-        assertTrue(Files.exists(snapshot.resolve("staged-content").resolve(sourceContentId + ".content")));
+        assertFalse(Files.exists(snapshot.resolve("staged-content").resolve(sourceContentId + ".content")));
-        assertTrue(Files.exists(snapshot.resolve("staged-content").resolve(sourceContentId + ".meta")));
+        assertFalse(Files.exists(snapshot.resolve("staged-content").resolve(sourceContentId + ".meta")));
         assertFalse(Files.exists(snapshot.resolve("staged-content").resolve(sourceContentId + ".owners")));
         System.out.println("snapshotPreservesSharedStatusContentWithoutStaleCredentialOwner...ok");
     }

pki/src/test/java/zeroecho/pki/impl/fs/FilesystemStagedContentStoreTest.java

@@ -146,13 +146,16 @@ final class FilesystemStagedContentStoreTest {
         FilesystemStagedContentStore store = new FilesystemStagedContentStore(directory, STORE_ID);
         DurableContentReference reference = stage(store, DurableContentReference.Lifecycle.OPERATION, "sign-input");
         DurableContentOwner authority = signingOwner("authority");
-        store.bindSigningOwnership(candidate -> candidate.equals(reference) ? Set.of(authority) : Set.of());
+        store.bindTransactionalOwnership(candidate -> candidate.equals(reference) ? Set.of(authority) : Set.of());
         assertThrows(IOException.class, () -> store.retireUnownedContent(reference));
         try (zeroecho.core.io.RepeatableContent content = store.openContent(reference)) {
             assertEquals(reference.length(), content.length().orElseThrow());
         }
         assertThrows(IllegalArgumentException.class,
                 () -> store.retainContent(reference, signingOwner("authority")));
+        assertThrows(IllegalArgumentException.class,
+                () -> store.retainContent(reference, new DurableContentOwner(
+                        DurableContentOwner.Category.STATUS_OBJECT_RECORD, "status-authority")));
         System.out.println("transactionalSigningOwnerPreventsGeneralRetirement...ok");
     }
 
@@ -161,8 +164,10 @@ final class FilesystemStagedContentStoreTest {
         System.out.println("sharedSigningPublicationReservationsAreReferenceCounted");
         FilesystemStagedContentStore store = new FilesystemStagedContentStore(directory, STORE_ID);
         DurableContentReference reference = stage(store, DurableContentReference.Lifecycle.OPERATION, "shared");
-        try (FilesystemStagedContentStore.SigningReservation first = store.reserveSigningPublication(reference);
+        try (FilesystemStagedContentStore.TransactionalReservation first =
-                FilesystemStagedContentStore.SigningReservation second = store.reserveSigningPublication(reference)) {
+                store.reserveTransactionalPublication(reference);
+                FilesystemStagedContentStore.TransactionalReservation second =
+                        store.reserveTransactionalPublication(reference)) {
             first.close();
             assertThrows(IOException.class, () -> store.retireUnownedContent(reference));
         }