C2a fix: a flattened record is marked by the document it stores; every writer decides its header by content

A flattened variant's HoldsIds is read off the document re-resolved against the post-batch base, not carried from the
variant or from the document the flatten was planned against. The conditional commit onto a file decides by the same
rule as the unconditional commit (ids in the records the file will hold, ids only this cache minted excepted, a marked
file staying marked). The directory manifest's header choice moves beside FormatFor. Pins the three-batch flatten
probe on the commit and flush paths, the conditional commit, the last id leaving, an id-bearing override that is not
the last, and a failed CultMesh write giving its ids back.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs

@@ -15,10 +15,8 @@ namespace GameCult.Caching.MessagePack;
 // A small hot manifest indexes the store; each record lives in one cold content-addressed page.
 public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
 {
-    private const string IndexedFormatVersion = "cultcache.store.v4.directory-content-addressed-pages";
-    // The manifest of a directory that holds element ids: a reader older than ids refuses it, because it would skip the id
-    // slots of a page and rewrite the elements without them.
-    private const string IndexedFormatVersionWithIds = "cultcache.store.v5.directory-content-addressed-pages";
+    private const string IndexedFormatVersion = CultPersistedStoreSnapshot.DirectoryFormatV4;
+    private const string IndexedFormatVersionWithIds = CultPersistedStoreSnapshot.DirectoryFormatV5;
     // An unleased load that has not settled after this many attempts throws.
     private const int UnleasedLoadAttempts = 5;
     private readonly FileInfo _manifestFile;
@@ -305,7 +303,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
     {
         var manifest = new CultPersistedStoreSnapshot
         {
-            FormatVersion = carriesElementIds ? IndexedFormatVersionWithIds : IndexedFormatVersion,
+            FormatVersion = CultPersistedStoreSnapshot.DirectoryFormatFor(carriesElementIds),
             SchemaCatalog = catalog,
             Records = index
         };

src/GameCult.Caching/CultCache.cs

@@ -65,6 +65,9 @@ namespace GameCult.Caching
         public const string FormatV1 = "cultcache.store.v1";
         public const string FormatV2 = "cultcache.store.v2";
         public const string FormatV3 = "cultcache.store.v3";
+        // The directory store's manifest formats: v5 when the directory holds an element id, for the reason v3 exists.
+        public const string DirectoryFormatV4 = "cultcache.store.v4.directory-content-addressed-pages";
+        public const string DirectoryFormatV5 = "cultcache.store.v5.directory-content-addressed-pages";
 
         public string FormatVersion { get; set; } = FormatV1;
 
@@ -74,6 +77,9 @@ namespace GameCult.Caching
         internal static string FormatFor(bool holdsIds, bool holdsVariants) =>
             holdsIds ? FormatV3 : holdsVariants ? FormatV2 : FormatV1;
 
+        // The directory store's header decision (it refuses variants): the same owner as FormatFor.
+        internal static string DirectoryFormatFor(bool holdsIds) => holdsIds ? DirectoryFormatV5 : DirectoryFormatV4;
+
         public CultSchemaCatalogEntry[] SchemaCatalog { get; set; } = Array.Empty<CultSchemaCatalogEntry>();
         public CultPersistedRecord[] Records { get; set; } = Array.Empty<CultPersistedRecord>();
     }
@@ -2743,7 +2749,13 @@ namespace GameCult.Caching
                 stack.RemoveAt(stack.Count - 1);
                 return done[key] = isVariant
                     ? variant.Resolved(document, minted ? new CultVariantDelta(delta.BaseKey, overrides) : delta, variant.IdsInMemoryOnly || (minted && source != null), holdsIds)
-                    : new CultStoredDocument(variant.Key, variant.StoredAt, variant.Descriptor, document) { Flattens = true, HoldsIds = variant.HoldsIds };
+                    : new CultStoredDocument(variant.Key, variant.StoredAt, variant.Descriptor, document)
+                    {
+                        Flattens = true,
+                        // The flattened record is the document it stores, resolved against the post-batch base: the flag is read
+                        // off that document, not carried from the variant or from the document the flatten was planned against.
+                        HoldsIds = CultElementIds.Plan(document, key, deterministic: source != null).HoldsIds
+                    };
             }
         }
 
@@ -3280,12 +3292,13 @@ namespace GameCult.Caching
             foreach (var entry in request.Upserts)
                 records[entry.Key.Value] = ToPersistedRecord(entry, SerializePayload);
 
-            // A file already marked stays marked: it may hold records this cache cannot read, and a rewrite must not shed the marker.
+            // One rule for every writer: the file holds ids when a record it will hold does. A record this commit replaces or removes
+            // no longer counts. Onto the file, only ids already persisted count (a load-minted id is on no disk), and a file already
+            // marked stays marked: it may hold records this cache cannot read, and a rewrite must not shed the marker.
             var replaced = request.Upserts.Concat(request.Deletes).Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
             var holdsIds = request.Upserts.Any(entry => entry.HoldsIds) ||
-                             (ontoDisk
-                                 ? string.Equals(disk.FormatVersion, CultPersistedStoreSnapshot.FormatV3, StringComparison.Ordinal)
-                                 : Entries.Values.Any(entry => entry.HoldsIds && !replaced.Contains(entry.Key.Value)));
+                           Entries.Values.Any(entry => entry.HoldsIds && !replaced.Contains(entry.Key.Value) && !(ontoDisk && entry.IdsInMemoryOnly)) ||
+                           (ontoDisk && string.Equals(disk.FormatVersion, CultPersistedStoreSnapshot.FormatV3, StringComparison.Ordinal));
             WriteSnapshot(records.Values, catalog.Concat(request.Upserts.Select(entry => entry.Descriptor.ToCatalogEntry())), holdsIds);
             if (!ontoDisk)
                 Cache?.IdsPersisted(records.Keys);

src/GameCult.Caching/GameCult.Caching.csproj

@@ -7,6 +7,7 @@
   <ItemGroup>
     <InternalsVisibleTo Include="GameCult.Caching.Tests" />
     <InternalsVisibleTo Include="GameCult.Mesh" />
+    <InternalsVisibleTo Include="GameCult.Caching.MessagePack" />
   </ItemGroup>
 
   <ItemGroup>

tests/GameCult.Caching.Tests/ElementIdTests.cs

@@ -770,6 +770,125 @@ namespace GameCult.Caching.Tests
             Assert.That(HeaderOf(withIds), Is.EqualTo("cultcache.store.v3"));
         }
 
+        // Soul's probe: the base gains ids in the batch that flattens the variant, then leaves. The flattened record is the
+        // document it stores, so its ids mark the file whatever the base did before or after.
+        [TestCase(false)]
+        [TestCase(true)]
+        public void AFlattenedVariantIsMarkedByTheDocumentItStoresNotByTheOneItWasPlannedAgainst(bool viaFlush)
+        {
+            var path = PathOf(viaFlush ? "flatten-flush.cc" : "flatten-commit.cc");
+            using (var cache = Open(path))
+            {
+                cache.Commit(batch =>
+                {
+                    batch.Upsert(typeof(IdDeck), EmptyDeck("base"), new CultRecordKey("base"));
+                    batch.UpsertVariant(new CultRecordKey("v"), new CultRecordKey("base"), new[] { cache.Override<IdDeck>(nameof(IdDeck.Name), "v") });
+                });
+                Assert.That(HeaderOf(path), Is.EqualTo("cultcache.store.v2"));
+
+                cache.Commit(batch =>
+                {
+                    batch.Upsert(typeof(IdDeck), Deck("base", reels: 1), new CultRecordKey("base"));
+                    batch.Flatten(new CultRecordKey("v"));
+                });
+                Assert.That(HeaderOf(path), Is.EqualTo("cultcache.store.v3"));
+
+                if (viaFlush)
+                {
+                    cache.Remove(new CultRecordKey("base"));
+                    cache.FlushAllBackingStores();
+                }
+                else
+                {
+                    Assert.That(cache.Commit(batch => batch.Remove(new CultRecordKey("base"))), Is.True);
+                }
+            }
+
+            var flattenedIds = AllIds(MessagePackSerializer.Deserialize<IdDeck>(DiskRecord(path, "v").Payload));
+            Assert.That(flattenedIds, Is.Not.Empty, "the flattened record holds the base's ids");
+            Assert.That(flattenedIds, Has.All.Not.Empty);
+            Assert.That(HeaderOf(path), Is.EqualTo("cultcache.store.v3"), "the file holds ids, so it is marked");
+        }
+
+        // A conditional commit lands onto the file as it is, and decides its header as every writer does: by the ids the file
+        // will hold. Ids on disk mark it whatever header the file carried; ids only this cache minted are on no disk.
+        [Test]
+        public void AConditionalCommitOntoAFileHoldingIdsMarksItAndOneOntoLoadMintedIdsDoesNot()
+        {
+            var held = PathOf("conditional-held.cc");
+            Type[] types = { typeof(IdDeck), typeof(PreCut2FixtureItem) };
+            using (var cache = OpenWith(held, false, types))
+                cache.Commit(batch => batch.Upsert(typeof(IdDeck), Deck("d"), new CultRecordKey("d")));
+            var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(held));
+            snapshot.FormatVersion = CultPersistedStoreSnapshot.FormatV1;
+            File.WriteAllBytes(held, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot));
+            using (var cache = OpenWith(held, false, types))
+            {
+                Assert.That(cache.Commit(batch =>
+                {
+                    batch.Expect(new CultRecordKey("plain"), null);
+                    batch.Upsert(typeof(PreCut2FixtureItem), new PreCut2FixtureItem { Name = "plain" }, new CultRecordKey("plain"));
+                }), Is.True);
+            }
+
+            Assert.That(HeaderOf(held), Is.EqualTo("cultcache.store.v3"), "the deck on disk holds ids");
+
+            var preId = PathOf("conditional-minted.cc");
+            WritePreIdStore(preId, null, ("a", "a", 1));
+            using (var cache = OpenWith(preId, false, types))
+            {
+                Assert.That(cache.Commit(batch =>
+                {
+                    batch.Expect(new CultRecordKey("plain"), null);
+                    batch.Upsert(typeof(PreCut2FixtureItem), new PreCut2FixtureItem { Name = "plain" }, new CultRecordKey("plain"));
+                }), Is.True);
+            }
+
+            Assert.That(HeaderOf(preId), Is.EqualTo("cultcache.store.v1"), "a's ids exist only in memory; the file holds none");
+        }
+
+        // The last id leaves the store in this commit: an unconditional commit writes the store it will hold, not the one it held.
+        [Test]
+        public void AnUnconditionalCommitThatRemovesOrReplacesTheLastIdWritesV1()
+        {
+            var removed = PathOf("last-removed.cc");
+            using (var cache = Open(removed))
+            {
+                cache.Commit(batch => batch.Upsert(typeof(IdDeck), Deck("d"), new CultRecordKey("d")));
+                Assert.That(HeaderOf(removed), Is.EqualTo("cultcache.store.v3"));
+                cache.Commit(batch => batch.Remove(new CultRecordKey("d")));
+            }
+
+            Assert.That(HeaderOf(removed), Is.EqualTo("cultcache.store.v1"));
+
+            var replaced = PathOf("last-replaced.cc");
+            using (var cache = Open(replaced))
+            {
+                cache.Commit(batch => batch.Upsert(typeof(IdDeck), Deck("d"), new CultRecordKey("d")));
+                cache.Commit(batch => batch.Upsert(typeof(IdDeck), EmptyDeck("d"), new CultRecordKey("d")));
+            }
+
+            Assert.That(HeaderOf(replaced), Is.EqualTo("cultcache.store.v1"));
+        }
+
+        // Any override holding an id marks the variant, wherever it sits among the overrides.
+        [Test]
+        public void AVariantWhoseIdBearingOverrideIsNotTheLastIsStillMarked()
+        {
+            var path = PathOf("variant-order.cc");
+            using (var cache = Open(path))
+            {
+                cache.Commit(batch =>
+                {
+                    batch.Upsert(typeof(IdDeck), EmptyDeck("base"), new CultRecordKey("base"));
+                    batch.UpsertVariant(new CultRecordKey("v"), new CultRecordKey("base"),
+                        new[] { cache.Override<IdDeck>(nameof(IdDeck.Reels), new List<IdReel> { new() { Label = "r" } }), cache.Override<IdDeck>(nameof(IdDeck.Name), "v") });
+                });
+            }
+
+            Assert.That(HeaderOf(path), Is.EqualTo("cultcache.store.v3"));
+        }
+
         [Test]
         public void AnElementObjectTwiceInOneListIsRefusedAtWriteAndNothingIsPersisted()
         {

tests/GameCult.Mesh.Tests/CultMeshSingleFileElementIdTests.cs

@@ -94,6 +94,18 @@ public sealed class CultMeshSingleFileElementIdTests
         File.ReadAllBytes(path).Should().Equal(before);
     }
 
+    [Test]
+    public void AFailedTypedWriteGivesTheMintedIdsBack()
+    {
+        var blocked = Path_("blocked.cc");
+        Directory.CreateDirectory(blocked);
+        var element = Descriptor();
+
+        Assert.That(() => CultMesh.WriteSingleFileDocument(blocked, Key, Publication(element)), Throws.Exception);
+
+        element.Id.Should().BeEmpty("a write that does not land gives its ids back");
+    }
+
     [Test]
     public void TypedReadRefusesWhatALoadRefusesAndMintsWhatALoadMints()
     {