fix(relations): final whole-branch review - all ten findings

Almost all of these live at the seams between tasks, which is why the
per-task reviews missed them.

1. BatchDelete bypassed relation enforcement entirely - store_.bulkDelete()
   with no canDelete, no cascade, no set_null, so any caller with ordinary
   write access could delete a restrict-protected parent by putting its id in
   a batch. Delete's block is now two shared helpers
   (checkRelationDeleteAllowed / performRelationCascade) that both handlers
   use. BatchDelete checks every id BEFORE deleting anything and refuses the
   whole batch, naming the blocking relation: its response carries only
   deleted_count, so a partial refusal would be indistinguishable from ids
   that never existed. The audit of every other delete/overwrite RPC also
   found DropCollection bypassing in BOTH directions - dropping the parent
   orphans every child, and dropping the CHILD leaves every reverse-index
   posting behind (drop_collection touches only the docs and vectors
   sub-dbs), so restrict then counts children that no longer exist and blocks
   the parent forever. Now refused, like SQL's DROP TABLE with an FK present.
   Still unfixed and reported: the TTL sweeper deletes a protected parent
   with no enforcement.

2. Delete was the one handler that could terminate the process. Its relation
   block caught only std::invalid_argument, but canDelete opens a ReadTxn and
   a cursor and mdb_check raises std::runtime_error - MDB_READERS_FULL and
   the EINVAL this project has had in production both land there, and an
   exception escaping a synchronous gRPC handler kills the service. The
   sibling read handlers all had the clause; this one did not.

3. The post-replay re-mirror pass no longer accumulates the touched set on
   the all-rows-fail path: Deferred holds {ds, qualified, bare, id} and phase
   3 re-reads the document, so MDB_MAP_FULL/ENOSPC/a poisoned handle can no
   longer turn a disk-full boot into an OOM-killed boot loop. The window is
   also byte-budgeted (64 MiB) as well as count-capped: 256 rows of this
   repo's ~2.9MB documents is ~750MB resident in the HAPPY path.

4. The pass ran BEFORE anything was armed, so it repaired the document and
   corrupted everything derived from it. recover() is line 88 of initialize()
   while the declarations are armed at ~199/~208, and put() reads
   indexed_fields/relations/unique_fields live - all three were empty. It now
   runs from initialize() after applyIndexDeclarations(), where
   backfillIntoDocStore() already sits. This was silently wrong rows on ANY
   install with a v2.9 index declared, relations or not. It also made the
   reverse-index half of relation_cascade.hpp's convergence claim false and
   phase 3's retry unreachable; both comments corrected, and the retry is now
   pinned by a test that fails five assertions without it.

5. T11's per-write Document deep copy is gated on a constraint actually being
   declared. It is yyjson_mut_val_mut_copy - O(size), a fresh allocation,
   inside the collection's unique_lock - and it was taken on every
   update-shaped write on every install, where with nothing declared the undo
   can never fire. Measured: 1.8us at 4KB, 7.1us at 64KB, 364us at 2.9MB. Be
   honest that end to end it is under 1% of update(); what it buys is
   lock-hold time and allocator churn, not throughput.

6. The cascade path is refused while the mirror is unhealthy or drifted.
   planCascade builds updatedDoc from LMDB and applyCascadeToMemory pushes it
   into MemoryStore, so when LMDB is legitimately behind - the state six read
   gates exist for - the cascade overwrites fresher data with older. Real
   data loss from a repair-shaped operation.

7. CreateRelation's bootstrap scan gets the same gate as CreateIndex(unique),
   for the same reason: indexing off LMDB while MemoryStore is ahead yields
   an incomplete index, and restrict then finds zero children for rows LMDB
   never received.

8. Canonicalised the child key - structurally, not as another spot fix. This
   was the THIRD occurrence of bare-vs-qualified in this file (v2.4.2 views,
   v2.4.5 collections): armRelationsForChild looked up the caller's raw
   string while boot arming used the canonical form, so a raw-gRPC caller
   naming "executions" could erase the real relation's armed state for the
   life of the process, logged only as "re-armed 0 relation(s)".
   RelationManager now canonicalises every name crossing its boundary and
   re-keys legacy rows in the store; CollectionConfigManager keys its cache
   canonically too, which was the other half of the mechanism
   (ConfigureCollection persisted under the raw string).

9. on_delete is validated, not coerced. Both duplicate parsers silently
   turned any unrecognised string into Restrict, which stopped failing safe
   when T12 made cascade/set_null destructive: an operator who types
   "Cascade" was told the relation was created and believed cascade was armed
   while restrict was. One parser now, returning optional; the RPC refuses
   and names the value, and a persisted bad value falls back to Restrict with
   an ERROR rather than being dropped (dropping removes protection).

10. client.hpp and database.proto (which ships in the -dev deb) both still
    told consumers cascade and set_null "behave as permit". Corrected, per
    policy, including the array rule and the grandchild refusal. The three
    load_test scripts no longer hardcode ROOT=/data/smartbotic-database - and
    the two that were already script-relative computed ROOT from a relative
    $0 AFTER cd-ing, so they failed unless invoked by absolute path; all five
    now resolve SCRIPT_DIR from BASH_SOURCE before any cd.

Tests: every fix verified by reverting it and re-running, except 2 and 7
(no harness can induce MDB_READERS_FULL in a live handler, or drift-but-
healthy on a live server - the same reason the CreateIndex(unique) gate is
parked untested). test_relation_enforcement 190 -> 252, test_relation_manager
12 -> 50; test_relation_index 43, test_subdb_identity 264,
test_dual_write_mirror 73, test_secondary_index_keys 41, test_document_store
all green, plus every other test binary. E2E from this worktree:
test_relations ALL PASSED, test_relations_client_e2e 33, test_policy_
enforcement 34, test_client_namespacing 44, test_views_multiproject ALL
PASSED, test_v24_tls_auth 4/4.

test_eviction aborts on a memory_priority assertion - verified PRE-EXISTING
by rebuilding it against branch-HEAD sources, not caused by this wave.

Report: .superpowers/sdd/2026-08-09-relations-v2.11.0/final-fix-report.md

client/include/smartbotic/database/client.hpp

@@ -794,8 +794,37 @@ public:
      * write access.
      *
      * `onDelete` is one of "restrict" (default), "cascade", "set_null",
-     * "no_action". Only restrict/no_action are enforced today - cascade and
-     * set_null are accepted and persisted but currently behave as permit.
+     * "no_action". ALL FOUR ARE ENFORCED as of v2.11.0. An unrecognised value
+     * is REFUSED, not coerced to "restrict".
+     *
+     *   - "restrict"  — the parent delete fails with FAILED_PRECONDITION while
+     *                   any child still references it; the message names the
+     *                   relation, the child count and up to five child ids.
+     *   - "cascade"   — DESTRUCTIVE. Deleting the parent deletes every child
+     *                   that references it, atomically with the parent. If the
+     *                   reference is an ARRAY of ids, the parent's id is pulled
+     *                   from the array and the child is KEPT (cascade and
+     *                   set_null collapse to the same behaviour there) - an
+     *                   array reference is a many-to-many edge, not the child's
+     *                   reason to exist. Refused if a child is itself protected
+     *                   by its own `restrict` relation; NOT recursive, so a
+     *                   grandchild under a cascade relation on the child is left
+     *                   dangling for `relations check` to find.
+     *   - "set_null"  — DESTRUCTIVE (a write, not a delete). The child's
+     *                   reference field is set to null and the child is kept.
+     *                   Array references behave as under "cascade".
+     *   - "no_action" — the delete proceeds and the reference is left dangling.
+     *                   The documented escape hatch.
+     *
+     * `validateOnWrite` is also enforced as of v2.11.0: a write to the child
+     * whose reference names a parent that does not exist is rejected with
+     * FAILED_PRECONDITION. An empty-string reference is a hard rejection.
+     *
+     * Per-collection `relations_enforced` (configureCollection) turns EVERY
+     * on_delete policy off for a collection, not just restrict - so a bulk
+     * import can be loaded without enforcement and checked afterwards with
+     * `relations check`. Turning it back on does NOT retroactively find
+     * references created while it was off.
      *
      * A cross-project relation (name/child/parent resolving to different
      * projects) is refused server-side: no LMDB transaction spans two

proto/database.proto

@@ -930,11 +930,24 @@ message RelationDefinition {
     string child_field = 3;         // dot-path on the child; may resolve to an array of ids
     string parent = 4;              // collection being referenced, project-qualified
     // "restrict" (default) | "cascade" | "set_null" | "no_action".
-    // Only restrict/no_action are enforced today (v2.11.0 T4/T6a); cascade
-    // and set_null are accepted and persisted but currently behave as
-    // permit (Task 12 makes them destructive).
+    //
+    // ALL FOUR ARE ENFORCED as of v2.11.0, and an unrecognised value is
+    // REFUSED by CreateRelation rather than coerced to "restrict".
+    //   restrict  - the parent delete fails FAILED_PRECONDITION while any
+    //               child still references it.
+    //   cascade   - DESTRUCTIVE: children referencing the parent are deleted
+    //               atomically with it. If child_field resolves to an ARRAY,
+    //               the parent id is pulled from the array and the child is
+    //               KEPT (cascade and set_null collapse there). Refused if a
+    //               child is itself protected by its own restrict relation.
+    //               Not recursive.
+    //   set_null  - DESTRUCTIVE: the child's reference field is set to null
+    //               and the child is kept. Arrays behave as under cascade.
+    //   no_action - the delete proceeds, the reference is left dangling.
     string on_delete = 5;
-    // Not yet enforced (Task 13). Accepted and persisted.
+    // ENFORCED as of v2.11.0: a write to the child naming a parent that does
+    // not exist is rejected with FAILED_PRECONDITION. An empty-string
+    // reference is a hard rejection.
     bool validate_on_write = 6;
     uint64 created_at = 7;
     uint64 updated_at = 8;

service/src/config/collection_config_manager.cpp

@@ -1,6 +1,7 @@
 #include "collection_config_manager.hpp"
 
 #include "../memory_store.hpp"
+#include "../project_addressing.hpp"
 
 #include <nlohmann/json.hpp>
 #include <spdlog/spdlog.h>
@@ -38,6 +39,15 @@ CollectionCfg fromJson(const nlohmann::json& j) {
 
 } // anonymous namespace
 
+std::string CollectionConfigManager::canonicalKey(const std::string& collection) {
+    if (collection.empty() || collection[0] == '_') return collection;
+    try {
+        return resolveCollection(collection).qualified;
+    } catch (const std::exception&) {
+        return collection;   // see the header: never throws on a hot path
+    }
+}
+
 CollectionConfigManager::CollectionConfigManager(MemoryStore& store) : store_(store) {}
 
 void CollectionConfigManager::loadFromStore() {
@@ -71,7 +81,7 @@ void CollectionConfigManager::loadFromStore() {
         if (result.documents.empty()) break;
         for (const auto& doc : result.documents) {
             if (doc.id.empty()) continue;
-            cache_[doc.id] = fromJson(doc.data());
+            cache_[canonicalKey(doc.id)] = fromJson(doc.data());
         }
         if (result.documents.size() < kPage) break;
         offset += kPage;
@@ -98,9 +108,13 @@ bool CollectionConfigManager::setConfig(const std::string& collection,
         return false;
     }
 
-    // Persist via upsert (document ID = collection name)
+    // Persist via upsert (document ID = the CANONICAL collection name, so a
+    // bare-named caller and a qualified-named caller write the same record
+    // rather than two half-configs - see canonicalKey()).
+    const std::string key = canonicalKey(collection);
+
     Document doc;
-    doc.id = collection;
+    doc.id = key;
     doc.set_data(toJson(cfg));
     try {
         store_.upsert(SYSTEM_COLLECTION, doc);
@@ -111,7 +125,7 @@ bool CollectionConfigManager::setConfig(const std::string& collection,
 
     {
         std::unique_lock<std::shared_mutex> wlock(cacheMutex_);
-        cache_[collection] = cfg;
+        cache_[key] = cfg;
     }
     spdlog::info("CollectionConfigManager: set '{}' timestamp_precision={}",
                  collection, cfg.timestampPrecision);
@@ -120,7 +134,7 @@ bool CollectionConfigManager::setConfig(const std::string& collection,
 
 CollectionCfg CollectionConfigManager::configFor(const std::string& collection) const {
     std::shared_lock<std::shared_mutex> lock(cacheMutex_);
-    auto it = cache_.find(collection);
+    auto it = cache_.find(canonicalKey(collection));
     if (it == cache_.end()) {
         return CollectionCfg{};  // default (ms)
     }
@@ -129,7 +143,7 @@ CollectionCfg CollectionConfigManager::configFor(const std::string& collection)
 
 bool CollectionConfigManager::hasExplicitConfig(const std::string& collection) const {
     std::shared_lock<std::shared_mutex> lock(cacheMutex_);
-    return cache_.contains(collection);
+    return cache_.contains(canonicalKey(collection));
 }
 
 } // namespace smartbotic::database

service/src/config/collection_config_manager.hpp

@@ -129,6 +129,27 @@ public:
     std::unordered_map<std::string, CollectionCfg> allConfigs() const;
 
 private:
+    /**
+     * v2.11.0 final review (finding 8) — the cache key for a USER collection
+     * is its canonical "<project>:<collection>" form, applied on every read
+     * and every write so the two cannot disagree.
+     *
+     * Without this, ConfigureCollection stored under whatever string the
+     * caller sent (`request->collection()`, raw) while the relation-arming
+     * path reads the canonical form - so `configureCollection("executions",
+     * relations_enforced=false)` wrote a key nothing ever reads, and the
+     * collection kept enforcing. Third occurrence of the bare-vs-qualified
+     * class in this repository (v2.4.2 views, v2.4.5 collections), so it is
+     * fixed at the keying layer rather than at each call site.
+     *
+     * `_`-prefixed system collections are left EXACTLY as given: they are
+     * global, not project-scoped, and `default:_views` does not exist - the
+     * same rule Client::qualify() follows since v2.7.0. Unparseable names are
+     * also returned unchanged, so a bad name reads back the default config
+     * instead of throwing on a hot path.
+     */
+    static std::string canonicalKey(const std::string& collection);
+
     MemoryStore& store_;
     mutable std::shared_mutex cacheMutex_;
     std::unordered_map<std::string, CollectionCfg> cache_;

service/src/database_grpc_impl.cpp

@@ -837,134 +837,27 @@ grpc::Status DatabaseGrpcImpl::Delete(
     // v2.11.0 T6a/T12 — referential integrity. After the access gate
     // (authorisation before integrity - a caller must not learn about
     // child counts on a collection they cannot read), before store_.remove()
-    // so a blocked delete never mutates anything. Only meaningful on the
-    // LMDB substrate, where the reverse index lives; if this project has no
-    // LMDB store for some reason, there is nothing to enforce against and
-    // the delete proceeds as before.
-    if (!request->collection().empty() && request->collection()[0] != '_') {
-        try {
-            const auto rc = smartbotic::database::resolveCollection(request->collection());
-            if (auto* ds = service_.docStore(rc.project)) {
-                if (auto* lmdb = dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds)) {
-                    // Named local: CollectionCfg is returned by value, and
-                    // RelationEnforcer binds relationsEnforced by reference —
-                    // binding straight to the temporary's subobject would dangle.
-                    const CollectionCfg cfg = config_manager_.configFor(request->collection());
-                    RelationEnforcer enforcer(relation_manager_, *lmdb, cfg.relationsEnforced);
-                    std::string err;
-                    if (!enforcer.canDelete(request->collection(), request->id(), err)) {
-                        return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, err);
-                    }
-
-                    // v2.11.0 T12 — cascade/set_null. restrict has already had
-                    // its say (canDelete, above); this is reached only when the
-                    // delete is otherwise permitted. relationsEnforced gates
-                    // this exactly like it gates restrict: disabled means every
-                    // on_delete policy is skipped, not just restrict, so a
-                    // collection with enforcement off keeps today's permissive
-                    // (dangling-reference) behaviour rather than half-enforcing.
-                    if (cfg.relationsEnforced) {
-                        bool hasDestructive = false;
-                        for (const auto& r : relation_manager_.relationsWithParent(request->collection())) {
-                            if (r.onDelete == smartbotic::database::OnDelete::Cascade ||
-                                r.onDelete == smartbotic::database::OnDelete::SetNull) {
-                                hasDestructive = true;
-                                break;
-                            }
-                        }
-                        if (hasDestructive) {
-                            bool parentExisted = false;
-                            // v2.11.0 T12 review (C2) — drive replication +
-                            // Subscribe events explicitly, once per mutation,
-                            // right after that mutation's MemoryStore apply.
-                            // See relation_cascade.hpp's file header for why
-                            // executeCascade() cannot go back through the
-                            // ordinary persistCallback_ path itself.
-                            auto notify = [this](const std::string& coll, const std::string& id,
-                                                 const std::optional<smartbotic::database::Document>& doc,
-                                                 smartbotic::database::EventType eventType) {
-                                service_.notifyReplicationAndEvents(coll, id, doc, eventType);
-                            };
-                            try {
-                                parentExisted = smartbotic::database::executeCascade(
-                                    relation_manager_, *lmdb, persistence_, store_, config_manager_,
-                                    request->collection(), request->id(), notify);
-                            } catch (const smartbotic::database::CascadeBlocked& e) {
-                                // v2.11.0 T12 review (I2) — a grandchild is
-                                // protected by its own restrict relation.
-                                // Nothing was written anywhere (this is thrown
-                                // from planCascade(), before writeCascadeWal()
-                                // ever runs) — same refusal shape restrict
-                                // itself uses.
-                                return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, e.what());
-                            } catch (const std::exception& e) {
-                                // v2.11.0 T12 review (I4, round 2 correction)
-                                // — WAL-first means a failure here can occur
-                                // AFTER the WAL entries for this cascade were
-                                // already durably fsynced (writeCascadeWal()
-                                // ran inside executeCascade() before
-                                // commitCascadeLmdb()/applyCascadeToMemory(),
-                                // either of which could be what actually
-                                // threw). This is NOT necessarily a no-op
-                                // failure. What the caller can and cannot
-                                // assume differs by WHICH of those two threw,
-                                // and neither is distinguishable from out
-                                // here:
-                                //   - commitCascadeLmdb() threw: LMDB was
-                                //     never committed (the WriteTxn aborts
-                                //     unwritten), so LMDB still reflects the
-                                //     pre-cascade state; MemoryStore is
-                                //     unchanged too (step 5 never ran). Reads
-                                //     see the pre-cascade state everywhere,
-                                //     for now — but the WAL entries are
-                                //     already durable, so the NEXT RESTART's
-                                //     replay applies the whole cascade
-                                //     regardless of this failure.
-                                //   - applyCascadeToMemory() threw partway:
-                                //     LMDB already committed (step 4
-                                //     succeeded) — reads being LMDB-first,
-                                //     the cascade IS ALREADY VISIBLE for
-                                //     collections read through LMDB, even
-                                //     though this call is reporting failure.
-                                //     MemoryStore may be only PARTIALLY
-                                //     applied (some mutations done, some
-                                //     not), and `notify` may have already
-                                //     fired for the mutations that did
-                                //     apply.
-                                // There is no compensating "un-write" of the
-                                // WAL entry in either case — replaying it
-                                // again is safe (idempotent), erasing it is
-                                // not. Check server logs and current
-                                // document state (LMDB, not just MemoryStore)
-                                // before retrying or assuming nothing
-                                // happened.
-                                spdlog::error(
-                                    "relations: cascade delete of '{}/{}' failed - the WAL entries for "
-                                    "this cascade may already be durable (will apply on next restart "
-                                    "regardless) and, if the failure was in the post-LMDB-commit step, "
-                                    "the mutation may ALREADY be visible via LMDB-first reads even "
-                                    "though this call is reporting failure: {}",
-                                    request->collection(), request->id(), e.what());
-                                return grpc::Status(grpc::StatusCode::INTERNAL,
-                                    "cascade delete failed - its WAL entries may already be durable "
-                                    "(will apply on the next restart regardless of this failure), and "
-                                    "if the failure occurred after the LMDB commit, the mutation may "
-                                    "ALREADY be visible via LMDB-first reads even though this call "
-                                    "failed; check server logs and current document state (not just "
-                                    "this response) before retrying: " + std::string(e.what()));
-                            }
-                            // executeCascade already deleted the parent (and its
-                            // vector, if any) atomically with every child
-                            // mutation — do not fall through to the plain
-                            // store_.remove() path below.
-                            response->set_deleted(parentExisted);
-                            return grpc::Status::OK;
-                        }
-                    }
-                }
-            }
-        } catch (const std::invalid_argument& e) {
-            return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, e.what());
+    // so a blocked delete never mutates anything.
+    //
+    // v2.11.0 final review (finding 1) — the body of this lives in
+    // checkRelationDeleteAllowed()/performRelationCascade() so that
+    // BatchDelete enforces identically instead of bypassing all of it.
+    if (auto st = checkRelationDeleteAllowed(request->collection(), request->id()); !st.ok()) {
+        return st;
+    }
+    {
+        bool handled = false;
+        bool parentExisted = false;
+        if (auto st = performRelationCascade(request->collection(), request->id(),
+                                            handled, parentExisted); !st.ok()) {
+            return st;
+        }
+        if (handled) {
+            // performRelationCascade already deleted the parent (and its
+            // vector, if any) atomically with every child mutation - do not
+            // fall through to the plain store_.remove() path below.
+            response->set_deleted(parentExisted);
+            return grpc::Status::OK;
         }
     }
 
@@ -1286,7 +1179,73 @@ grpc::Status DatabaseGrpcImpl::BatchDelete(
             "cannot write to view '" + request->collection() + "': views are read-only");
     }
     std::vector<std::string> ids(request->ids().begin(), request->ids().end());
-    uint64_t deleted = store_.bulkDelete(request->collection(), ids);
+
+    // v2.11.0 final review (finding 1) — REFERENTIAL INTEGRITY. This handler
+    // used to call store_.bulkDelete() straight through: no canDelete, no
+    // cascade, no set_null. Any caller with ordinary write access could delete
+    // a restrict-protected parent just by putting its id in a batch, defeating
+    // the feature through a sibling RPC. Same enforcement Delete uses, via the
+    // same two helpers, so the two cannot drift.
+    //
+    // ALL-OR-NOTHING on the restrict check, and deliberately so: BatchDelete's
+    // response carries only deleted_count, with no per-id error channel, so a
+    // partial refusal would be indistinguishable from ids that simply did not
+    // exist. Checking every id BEFORE deleting anything also preserves Delete's
+    // property that a blocked delete never mutates anything. The refusal names
+    // the blocking relation for each blocked id, since that is what the caller
+    // needs in order to act.
+    {
+        std::string blocked;
+        size_t blockedCount = 0;
+        for (const auto& id : ids) {
+            auto st = checkRelationDeleteAllowed(request->collection(), id);
+            if (st.ok()) continue;
+            if (st.error_code() != grpc::StatusCode::FAILED_PRECONDITION) {
+                return st;   // INVALID_ARGUMENT / INTERNAL - surface as-is
+            }
+            ++blockedCount;
+            // Cap the message: a batch of 10k blocked ids must not produce a
+            // 10k-entry error string. The count is exact regardless.
+            if (blockedCount <= 5) {
+                if (!blocked.empty()) blocked += "; ";
+                blocked += st.error_message();
+            }
+        }
+        if (blockedCount > 0) {
+            return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
+                "batch delete refused: " + std::to_string(blockedCount) + " of " +
+                std::to_string(ids.size()) + " document(s) are protected by a relation, "
+                "and nothing was deleted (BatchDelete has no per-id error channel, so it "
+                "is all-or-nothing). " +
+                (blockedCount > 5 ? "First five: " : "") + blocked);
+        }
+    }
+
+    uint64_t deleted = 0;
+    std::vector<std::string> plain;      // ids with no destructive relation
+    plain.reserve(ids.size());
+    for (const auto& id : ids) {
+        bool handled = false;
+        bool parentExisted = false;
+        // Per id, because a cascade is one atomic LMDB transaction per parent
+        // (see relation_cascade.hpp) - there is no batched form of it. Any
+        // failure aborts the rest of the batch rather than continuing: after a
+        // cascade fault the WAL may already hold durable entries (see
+        // performRelationCascade's INTERNAL branch), and pressing on would make
+        // the reported count meaningless.
+        if (auto st = performRelationCascade(request->collection(), id,
+                                            handled, parentExisted); !st.ok()) {
+            return st;
+        }
+        if (handled) {
+            if (parentExisted) ++deleted;
+        } else {
+            plain.push_back(id);
+        }
+    }
+    if (!plain.empty()) {
+        deleted += store_.bulkDelete(request->collection(), plain);
+    }
     response->set_deleted_count(deleted);
     return grpc::Status::OK;
 }
@@ -1828,6 +1787,55 @@ grpc::Status DatabaseGrpcImpl::DropCollection(
     if (store_.pressure() == MemoryPressure::Emergency) {
         return memoryEmergencyStatus("DropCollection", store_);
     }
+
+    // v2.11.0 final review (finding 1, from the "audit every other RPC that can
+    // delete a document" sweep) — DROPPING A COLLECTION IS A DELETE OF EVERY
+    // DOCUMENT IN IT, and this handler enforced nothing. Both directions were
+    // broken, and neither logged anything:
+    //
+    //   - Dropping the PARENT collection destroyed every restrict-protected
+    //     parent at once, leaving every child pointing at nothing - the exact
+    //     outcome the feature exists to prevent, reachable with one RPC.
+    //   - Dropping the CHILD collection is worse in a quieter way:
+    //     LmdbDocumentStore::drop_collection() drops the docs and vectors
+    //     sub-dbs ONLY, so every reverse-index posting survives. `restrict`
+    //     then counts children that no longer exist and blocks the parent's
+    //     delete forever, with `relations check` unable to explain it because
+    //     the child collection is gone.
+    //
+    // Refused rather than cascaded: this is what SQL's `DROP TABLE` does with
+    // a foreign key present (RESTRICT is the default), the remedy is one
+    // command (`DropRelation`), and silently dropping the relation on the
+    // operator's behalf would destroy a declaration they may want back.
+    //
+    // ⚠ Deliberately NOT gated on relationsEnforced. That flag is the escape
+    // hatch for "let this delete leave a dangling reference"; the child-side
+    // consequence here is a stale index that corrupts future enforcement for a
+    // DIFFERENT collection, which is a storage-consistency problem rather than
+    // an enforcement policy one.
+    {
+        std::vector<std::string> involved;
+        for (const auto& r : relation_manager_.relationsWithParent(request->name())) {
+            involved.push_back(r.name + " (as parent, child " + r.child + "." + r.childField + ")");
+        }
+        for (const auto& r : relation_manager_.relationsWithChild(request->name())) {
+            involved.push_back(r.name + " (as child, parent " + r.parent + ")");
+        }
+        if (!involved.empty()) {
+            std::string list;
+            for (const auto& s : involved) {
+                if (!list.empty()) list += "; ";
+                list += s;
+            }
+            return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
+                "cannot drop collection '" + request->name() + "': it participates in " +
+                std::to_string(involved.size()) + " declared relation(s) - " + list +
+                ". Dropping it would either orphan every child row or leave reverse-index "
+                "postings for rows that no longer exist (which would block the parent's "
+                "deletes permanently). Drop the relation(s) first.");
+        }
+    }
+
     bool dropped = store_.dropCollection(request->name());
     response->set_dropped(dropped);
 
@@ -2812,22 +2820,14 @@ grpc::Status DatabaseGrpcImpl::GetViewInfo(
 // (or open, when nothing anywhere is secured — the pre-2.7.0 world).
 
 namespace {
-std::string relationOnDeleteToString(OnDelete v) {
-    switch (v) {
-        case OnDelete::Restrict: return "restrict";
-        case OnDelete::Cascade: return "cascade";
-        case OnDelete::SetNull: return "set_null";
-        case OnDelete::NoAction: return "no_action";
-    }
-    return "restrict";
-}
-
-OnDelete relationOnDeleteFromString(const std::string& s) {
-    if (s == "cascade") return OnDelete::Cascade;
-    if (s == "set_null") return OnDelete::SetNull;
-    if (s == "no_action") return OnDelete::NoAction;
-    return OnDelete::Restrict;
-}
+// v2.11.0 final review (finding 9) — the local relationOnDeleteToString /
+// relationOnDeleteFromString pair is GONE. Both were duplicates of
+// relation_manager's own, and the FromString half silently coerced any
+// unrecognised value to Restrict. There is now exactly one parser
+// (smartbotic::database::parseOnDelete, which returns nullopt instead of
+// guessing) and one renderer (onDeleteToString).
+using smartbotic::database::onDeleteToString;
+using smartbotic::database::parseOnDelete;
 
 pb::RelationDefinition relationToProto(const RelationInfo& r) {
     pb::RelationDefinition out;
@@ -2835,7 +2835,7 @@ pb::RelationDefinition relationToProto(const RelationInfo& r) {
     out.set_child(r.child);
     out.set_child_field(r.childField);
     out.set_parent(r.parent);
-    out.set_on_delete(relationOnDeleteToString(r.onDelete));
+    out.set_on_delete(onDeleteToString(r.onDelete));
     out.set_validate_on_write(r.validateOnWrite);
     out.set_created_at(r.createdAt);
     out.set_updated_at(r.updatedAt);
@@ -2843,6 +2843,166 @@ pb::RelationDefinition relationToProto(const RelationInfo& r) {
 }
 } // namespace
 
+// -------------------------------------------------------------------------
+// v2.11.0 final review (finding 1) — shared delete-side relation enforcement.
+// See the declarations in database_grpc_impl.hpp for why this is shared rather
+// than living inline in Delete().
+// -------------------------------------------------------------------------
+
+grpc::Status DatabaseGrpcImpl::checkRelationDeleteAllowed(const std::string& collection,
+                                                          const std::string& id) {
+    // Only meaningful on the LMDB substrate, where the reverse index lives; if
+    // this project has no LMDB store for some reason there is nothing to
+    // enforce against and the delete proceeds as before.
+    if (collection.empty() || collection[0] == '_') return grpc::Status::OK;
+    try {
+        const auto rc = smartbotic::database::resolveCollection(collection);
+        auto* ds = service_.docStore(rc.project);
+        if (ds == nullptr) return grpc::Status::OK;
+        auto* lmdb = dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds);
+        if (lmdb == nullptr) return grpc::Status::OK;
+
+        // Named local: CollectionCfg is returned by value, and RelationEnforcer
+        // binds relationsEnforced by reference — binding straight to the
+        // temporary's subobject would dangle.
+        const CollectionCfg cfg = config_manager_.configFor(collection);
+        RelationEnforcer enforcer(relation_manager_, *lmdb, cfg.relationsEnforced);
+        std::string err;
+        if (!enforcer.canDelete(collection, id, err)) {
+            return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, err);
+        }
+        return grpc::Status::OK;
+    } catch (const std::invalid_argument& e) {
+        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, e.what());
+    } catch (const std::exception& e) {
+        // v2.11.0 final review (finding 2) — MANDATORY, not defensive. This
+        // path opens an LMDB ReadTxn and a cursor (canDelete ->
+        // findRelationBlocks -> lookupRelationCounts ->
+        // relation_index_child_count), and mdb_check/throw_mdb raise
+        // std::runtime_error - MDB_READERS_FULL and the EINVAL this project has
+        // already had in production both land here. Delete() previously caught
+        // only std::invalid_argument, and an exception escaping a synchronous
+        // gRPC handler TERMINATES THE PROCESS: a full reader table would have
+        // taken the service down rather than failing one call. DescribeDelete,
+        // CheckRelation, CreateIndex and Exists all catch std::exception on the
+        // same read path; this is the clause that was missing.
+        spdlog::error("relations: delete enforcement failed for '{}/{}': {}",
+                      collection, id, e.what());
+        return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
+    }
+}
+
+grpc::Status DatabaseGrpcImpl::performRelationCascade(const std::string& collection,
+                                                      const std::string& id,
+                                                      bool& handled,
+                                                      bool& parentExisted) {
+    handled = false;
+    parentExisted = false;
+    if (collection.empty() || collection[0] == '_') return grpc::Status::OK;
+    try {
+        const auto rc = smartbotic::database::resolveCollection(collection);
+        auto* ds = service_.docStore(rc.project);
+        if (ds == nullptr) return grpc::Status::OK;
+        auto* lmdb = dynamic_cast<smartbotic::db::storage::LmdbDocumentStore*>(ds);
+        if (lmdb == nullptr) return grpc::Status::OK;
+
+        // v2.11.0 T12 — cascade/set_null. restrict has already had its say
+        // (checkRelationDeleteAllowed, called first by every caller); this is
+        // reached only when the delete is otherwise permitted.
+        // relationsEnforced gates this exactly like it gates restrict:
+        // disabled means every on_delete policy is skipped, not just restrict,
+        // so a collection with enforcement off keeps today's permissive
+        // (dangling-reference) behaviour rather than half-enforcing.
+        const CollectionCfg cfg = config_manager_.configFor(collection);
+        if (!cfg.relationsEnforced) return grpc::Status::OK;
+
+        bool hasDestructive = false;
+        for (const auto& r : relation_manager_.relationsWithParent(collection)) {
+            if (r.onDelete == smartbotic::database::OnDelete::Cascade ||
+                r.onDelete == smartbotic::database::OnDelete::SetNull) {
+                hasDestructive = true;
+                break;
+            }
+        }
+        if (!hasDestructive) return grpc::Status::OK;
+
+        // v2.11.0 T12 review (C2) — drive replication + Subscribe events
+        // explicitly, once per mutation, right after that mutation's
+        // MemoryStore apply. See relation_cascade.hpp's file header for why
+        // executeCascade() cannot go back through the ordinary
+        // persistCallback_ path itself.
+        auto notify = [this](const std::string& coll, const std::string& docId,
+                             const std::optional<smartbotic::database::Document>& doc,
+                             smartbotic::database::EventType eventType) {
+            service_.notifyReplicationAndEvents(coll, docId, doc, eventType);
+        };
+        try {
+            parentExisted = smartbotic::database::executeCascade(
+                relation_manager_, *lmdb, persistence_, store_, config_manager_,
+                collection, id, notify);
+            handled = true;
+            return grpc::Status::OK;
+        } catch (const smartbotic::database::CascadeBlocked& e) {
+            // v2.11.0 T12 review (I2) — a grandchild is protected by its own
+            // restrict relation. Also (final review, finding 6) the
+            // mirror-unhealthy/drifted refusal. Nothing was written anywhere in
+            // either case: both are thrown before writeCascadeWal() ever runs —
+            // same refusal shape restrict itself uses.
+            return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION, e.what());
+        } catch (const std::exception& e) {
+            // v2.11.0 T12 review (I4, round 2 correction) — WAL-first means a
+            // failure here can occur AFTER the WAL entries for this cascade
+            // were already durably fsynced (writeCascadeWal() ran inside
+            // executeCascade() before commitCascadeLmdb()/
+            // applyCascadeToMemory(), either of which could be what actually
+            // threw). This is NOT necessarily a no-op failure. What the caller
+            // can and cannot assume differs by WHICH of those two threw, and
+            // neither is distinguishable from out here:
+            //   - commitCascadeLmdb() threw: LMDB was never committed (the
+            //     WriteTxn aborts unwritten), so LMDB still reflects the
+            //     pre-cascade state; MemoryStore is unchanged too (step 5 never
+            //     ran). Reads see the pre-cascade state everywhere, for now —
+            //     but the WAL entries are already durable, so the NEXT
+            //     RESTART's replay applies the whole cascade regardless of this
+            //     failure.
+            //   - applyCascadeToMemory() threw partway: LMDB already committed
+            //     (step 4 succeeded) — reads being LMDB-first, the cascade IS
+            //     ALREADY VISIBLE for collections read through LMDB, even
+            //     though this call is reporting failure. MemoryStore may be
+            //     only PARTIALLY applied (some mutations done, some not), and
+            //     `notify` may have already fired for the mutations that did
+            //     apply.
+            // There is no compensating "un-write" of the WAL entry in either
+            // case — replaying it again is safe (idempotent), erasing it is
+            // not. Check server logs and current document state (LMDB, not
+            // just MemoryStore) before retrying or assuming nothing happened.
+            spdlog::error(
+                "relations: cascade delete of '{}/{}' failed - the WAL entries for "
+                "this cascade may already be durable (will apply on next restart "
+                "regardless) and, if the failure was in the post-LMDB-commit step, "
+                "the mutation may ALREADY be visible via LMDB-first reads even "
+                "though this call is reporting failure: {}",
+                collection, id, e.what());
+            return grpc::Status(grpc::StatusCode::INTERNAL,
+                "cascade delete failed - its WAL entries may already be durable "
+                "(will apply on the next restart regardless of this failure), and "
+                "if the failure occurred after the LMDB commit, the mutation may "
+                "ALREADY be visible via LMDB-first reads even though this call "
+                "failed; check server logs and current document state (not just "
+                "this response) before retrying: " + std::string(e.what()));
+        }
+    } catch (const std::invalid_argument& e) {
+        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, e.what());
+    } catch (const std::exception& e) {
+        // finding 2 — same reasoning as checkRelationDeleteAllowed's clause:
+        // relationsWithParent()/configFor()/resolveCollection() must not be
+        // able to terminate the process.
+        spdlog::error("relations: cascade planning failed for '{}/{}': {}",
+                      collection, id, e.what());
+        return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
+    }
+}
+
 void DatabaseGrpcImpl::armRelationsForChild(const std::string& childQualified) {
     try {
         const auto rc = smartbotic::database::resolveCollection(childQualified);
@@ -2867,10 +3027,24 @@ void DatabaseGrpcImpl::armRelationsForChild(const std::string& childQualified) {
         // itself (see RelationRef's comment). ConfigureCollection re-arms
         // after flipping the flag, so this only ever goes stale between that
         // RPC's write and its own re-arm call - never observably.
-        const bool enforced = config_manager_.configFor(childQualified).relationsEnforced;
+        //
+        // v2.11.0 final review (finding 8) — rc.qualified, NOT the caller's
+        // raw string. Boot arming (applyRelationDeclarations) uses the
+        // canonical `project + ":" + collection`, and set_relations() stores
+        // under the BARE name either way, so a raw-gRPC caller naming
+        // "executions" instead of "default:executions" used to look up zero
+        // relations here and then set_relations(bare, {}) on the SAME map
+        // entry the canonical arming had filled - disarming both the reverse
+        // index and validate_on_write for the life of the process, logged only
+        // as "re-armed 0 relation(s)", and silently repaired by a restart.
+        // RelationManager now canonicalises internally too (belt and braces:
+        // the lookup is correct even if a future caller reaches it raw), and
+        // CollectionConfigManager keys its cache canonically for the same
+        // reason.
+        const bool enforced = config_manager_.configFor(rc.qualified).relationsEnforced;
 
         std::vector<smartbotic::db::storage::RelationRef> refs;
-        for (const auto& rel : relation_manager_.relationsWithChild(childQualified)) {
+        for (const auto& rel : relation_manager_.relationsWithChild(rc.qualified)) {
             const auto rn = smartbotic::database::resolveCollection(rel.name);
             // v2.11.0 T13 — parent is resolved to its BARE collection name,
             // same as name/childField above: relations never cross projects
@@ -2909,6 +3083,28 @@ grpc::Status DatabaseGrpcImpl::CreateRelation(
             std::to_string(store_.pressurePercent()) + "%); retry after backoff");
         return grpc::Status::OK;
     }
+    // v2.11.0 final review (finding 7) — the same gate CreateIndex(unique)
+    // carries, for the same reason. The bootstrap scan below
+    // (build_relation_index) builds the reverse index by walking LMDB, and
+    // while the mirror is unhealthy or has drifted MemoryStore is
+    // legitimately AHEAD of LMDB - that is what the fallback is for. Indexing
+    // off LMDB in that state produces a silently incomplete index, and
+    // `restrict` then reports zero children for every row LMDB never
+    // received, permitting exactly the parent deletes this relation is being
+    // declared to prevent. Refused up front rather than half-built: the
+    // declaration would look successful and `rows_indexed` would even report
+    // a plausible number.
+    if (!service_.mirrorHealthy() || service_.mirrorDriftCount() != 0) {
+        response->set_success(false);
+        response->set_error("cannot declare a relation while the LMDB mirror is unhealthy "
+                            "or has drifted: the bootstrap scan that indexes existing rows "
+                            "reads LMDB only, and MemoryStore may be ahead of it right now, "
+                            "so the reverse index would come out incomplete and `restrict` "
+                            "would permit deletes it should refuse. Check the mirror drift "
+                            "count in the service log; a restart clears drift once the "
+                            "underlying cause is fixed.");
+        return grpc::Status::OK;
+    }
 
     try {
         RelationInfo r;
@@ -2916,7 +3112,24 @@ grpc::Status DatabaseGrpcImpl::CreateRelation(
         r.child = request->child();
         r.childField = request->child_field();
         r.parent = request->parent();
-        r.onDelete = relationOnDeleteFromString(request->on_delete());
+        // v2.11.0 final review (finding 9) — VALIDATE, do not coerce. An
+        // unrecognised on_delete used to become Restrict silently, which
+        // stopped being a safe default the moment T12 made cascade/set_null
+        // genuinely destructive: an operator who typed "Cascade" was told the
+        // relation was created and believed cascade was armed while restrict
+        // was, and only discovered it when the deletes they expected to
+        // cascade started refusing. Empty still means the documented default.
+        if (request->on_delete().empty()) {
+            r.onDelete = OnDelete::Restrict;
+        } else if (auto parsed = parseOnDelete(request->on_delete())) {
+            r.onDelete = *parsed;
+        } else {
+            response->set_success(false);
+            response->set_error("on_delete must be one of " +
+                                std::string(smartbotic::database::kOnDeleteValues) +
+                                " (got '" + request->on_delete() + "')");
+            return grpc::Status::OK;
+        }
         r.validateOnWrite = request->validate_on_write();
 
         // Cross-project / malformed-name validation (no LMDB transaction
@@ -3112,7 +3325,7 @@ grpc::Status DatabaseGrpcImpl::DescribeDelete(
             pbImpact->set_relation(imp.relation);
             pbImpact->set_child_collection(imp.childCollection);
             pbImpact->set_child_field(imp.childField);
-            pbImpact->set_on_delete(relationOnDeleteToString(imp.onDelete));
+            pbImpact->set_on_delete(onDeleteToString(imp.onDelete));
             pbImpact->set_child_count(imp.childCount);
             for (const auto& id : imp.sampleChildIds) {
                 pbImpact->add_sample_child_ids(id);

service/src/database_grpc_impl.hpp

@@ -459,6 +459,41 @@ private:
     // since the declaration itself already succeeded and persisted.
     void armRelationsForChild(const std::string& childQualified);
 
+    // -------------------------------------------------------------------
+    // v2.11.0 final review (finding 1) — referential integrity for deletes,
+    // extracted out of Delete() so BATCHDELETE SHARES IT.
+    //
+    // BatchDelete used to call store_.bulkDelete() directly: no canDelete, no
+    // cascade, no set_null. Any caller with ordinary write access could delete
+    // a restrict-protected parent simply by putting its id in a batch, which
+    // defeats the entire feature through a sibling RPC. Exactly the miss class
+    // the v2.7.0 policy audit caught for Upsert/Batch*/Subscribe - found then
+    // by auditing every handler rather than by reading one.
+    //
+    // Both helpers are no-ops for `_`-prefixed system collections and for
+    // projects with no LMDB substrate (the reverse index lives in LMDB), and
+    // both map an LMDB fault to a non-OK status rather than letting it escape
+    // - see finding 2.
+    // -------------------------------------------------------------------
+
+    // Restrict/no_action only: non-OK means this delete must be refused, and
+    // the message names every blocking relation, its child count and sample
+    // child ids. Reads nothing but the reverse index's cursor counts, so it is
+    // cheap enough to run over a whole batch before touching anything.
+    grpc::Status checkRelationDeleteAllowed(const std::string& collection,
+                                            const std::string& id);
+
+    // Cascade/set_null. `handled` false means this collection declares no
+    // destructive relation and the caller must perform the ordinary delete
+    // itself; true means the delete (parent included) has ALREADY been
+    // performed atomically and `parentExisted` says whether there was a
+    // parent to remove. Assumes checkRelationDeleteAllowed() has already
+    // permitted the delete, exactly as Delete's original ordering did.
+    grpc::Status performRelationCascade(const std::string& collection,
+                                        const std::string& id,
+                                        bool& handled,
+                                        bool& parentExisted);
+
     DatabaseService& service_;
     MemoryStore& store_;
     PersistenceManager& persistence_;

service/src/database_service.cpp

@@ -207,6 +207,30 @@ bool DatabaseService::initialize() {
         // with nothing logged.
         applyIndexDeclarations();
 
+        // v2.11.0 final review (finding 4) — THE POST-REPLAY RE-MIRROR PASS
+        // RUNS HERE, and the position is load-bearing.
+        //
+        // PersistenceManager::recover() (above, before any of the arming
+        // steps) only collects the id list. The pass writes documents through
+        // LmdbDocumentStore::put(), which reads indexed_fields(),
+        // unique_fields() and relations() LIVE and maintains every index
+        // inside the document's own write transaction. Run from inside
+        // recover(), all three maps were still empty, so it repaired the
+        // document and left every posting derived from it stale: an
+        // index-served query then returns the row under its OLD value and
+        // misses it under its new one - silently wrong rows, which is exactly
+        // what putting maintainIndexes() inside the transaction exists to
+        // prevent, and which bit any install with a v2.9 index declared
+        // whether or not it uses relations. It also made the reverse-index
+        // half of relation_cascade.hpp's convergence claim false, and made
+        // UniqueViolation (and therefore the pass's own retry phase)
+        // unreachable.
+        //
+        // Must be after applyRelationDeclarations() AND
+        // applyIndexDeclarations(); backfillIntoDocStore() below already sits
+        // at this point for the same reason. Never throws.
+        persistence_->runPendingRemirror(*store_, recovery_outcome_);
+
         // Run migrations if enabled
         if (config_.migrations.enabled && !config_.migrations.directory.empty()) {
             if (!runMigrations()) {

service/src/memory_store.cpp

@@ -503,10 +503,23 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
     // v2.11.0 T11 — snapshot enough of the pre-write state to undo, in case
     // the mirror rejects this as a UniqueViolation. Captured before anything
     // is mutated: the document itself, and whether/what vector it held.
-    const Document original = it->second;
+    // ⚠ v2.11.0 final review (finding 5) — CONDITIONAL. Document's copy
+    // constructor is yyjson_mut_val_mut_copy: O(document size), a fresh
+    // yyjson_mut_doc, taken here inside the collection's unique_lock. T11 took
+    // it on EVERY update-shaped write on EVERY install, but the undo can only
+    // ever run when the mirror can reject the write - i.e. when this collection
+    // has a declared unique field or a validate_on_write relation. Where none
+    // is declared the copy was pure cost on the path v2.8.0 spent a release
+    // making cheaper. needsUndoSnapshot() answers the common "nothing declared
+    // anywhere" case with one relaxed atomic load, and fails SAFE (true) on any
+    // doubt.
+    const bool wantUndo = needsUndoSnapshot(collection);
+    const Document original = wantUndo ? it->second : Document{};
     std::optional<std::vector<float>> originalVec;
-    if (auto vit = coll->vectors.find(id); vit != coll->vectors.end()) {
-        originalVec = vit->second;
+    if (wantUndo) {
+        if (auto vit = coll->vectors.find(id); vit != coll->vectors.end()) {
+            originalVec = vit->second;
+        }
     }
 
     // Remove from old expiration index
@@ -571,6 +584,20 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
     // vector, and the expiration index back exactly as `original` had them,
     // so a rejected update leaves the row as if the update never happened.
     mirrorDocOrUndo(collection, id, updated, EventType::UPDATE, [&]() {
+        // finding 5 — the snapshot is conditional, so the undo must be too.
+        // With no rejecting constraint declared, `original` is an EMPTY
+        // Document and restoring from it would DESTROY the row, so refuse and
+        // leave the mutation in place (the pre-T11 behaviour, and the lesser
+        // harm). Unreachable in practice: put() cannot raise
+        // UniqueViolation/MissingParentReference without a declaration, which
+        // is the same condition needsUndoSnapshot() tests.
+        if (!wantUndo) {
+            spdlog::error("mirror rejected a write to '{}/{}' but no undo snapshot "
+                          "was taken (no unique field or validate_on_write relation "
+                          "is declared for this collection) - the in-memory row is "
+                          "left as written", collection, id);
+            return;
+        }
         if (updated.expiresAt > 0) {
             removeFromExpirationIndex(*coll, id, updated.expiresAt);
         }
@@ -648,11 +675,22 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
 
     // v2.11.0 T11 — snapshot pre-write state for undo before either branch
     // mutates anything. Only meaningful on the update branch (isInsert=false
-    // undo just erases what it added), but cheap enough to always capture.
+    // undo just erases what it added).
+    // ⚠ v2.11.0 final review (finding 5) — CONDITIONAL. Document's copy
+    // constructor is yyjson_mut_val_mut_copy: O(document size), a fresh
+    // yyjson_mut_doc, taken here inside the collection's unique_lock. T11 took
+    // it on EVERY update-shaped write on EVERY install, but the undo can only
+    // ever run when the mirror can reject the write - i.e. when this collection
+    // has a declared unique field or a validate_on_write relation. Where none
+    // is declared the copy was pure cost on the path v2.8.0 spent a release
+    // making cheaper. needsUndoSnapshot() answers the common "nothing declared
+    // anywhere" case with one relaxed atomic load, and fails SAFE (true) on any
+    // doubt.
+    const bool wantUndo = needsUndoSnapshot(collection);
     const bool hadExisting = it != coll->documents.end();
-    const Document originalDoc = hadExisting ? it->second : Document{};
+    const Document originalDoc = (wantUndo && hadExisting) ? it->second : Document{};
     std::optional<std::vector<float>> originalVec;
-    if (hadExisting) {
+    if (wantUndo && hadExisting) {
         if (auto vit = coll->vectors.find(docId); vit != coll->vectors.end()) {
             originalVec = vit->second;
         }
@@ -719,6 +757,17 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
     // vector/expiration index) to the pre-upsert snapshot.
     mirrorDocOrUndo(collection, docId, doc,
                     isInsert ? EventType::INSERT : EventType::UPDATE, [&]() {
+        // finding 5 — the INSERT branch's undo needs no snapshot (it just
+        // erases what it added), so it is always safe. The UPDATE branch
+        // restores from `originalDoc` and must refuse when that was never
+        // captured - see the note on the snapshot above.
+        if (!isInsert && !wantUndo) {
+            spdlog::error("mirror rejected an upsert of '{}/{}' but no undo snapshot "
+                          "was taken (no unique field or validate_on_write relation "
+                          "is declared for this collection) - the in-memory row is "
+                          "left as written", collection, docId);
+            return;
+        }
         if (doc.expiresAt > 0) {
             removeFromExpirationIndex(*coll, docId, doc.expiresAt);
         }
@@ -874,13 +923,15 @@ bool MemoryStore::unloadDocument(const std::string& collection, const std::strin
 
 MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
     const std::vector<std::pair<std::string, std::string>>& docs,
-    size_t chunkSize) {
+    size_t chunkSize,
+    uint64_t chunkBytes) {
     // v2.11.0 T12 round-4 — see the header comment. This function must never
-    // throw: it runs inside PersistenceManager::recover(), and an escaping
-    // exception there aborts startup permanently.
+    // throw: it runs on the boot path, and an escaping exception there aborts
+    // startup permanently.
     RemirrorBatchResult result;
     if (docs.empty()) return result;
     if (chunkSize == 0) chunkSize = 1;
+    if (chunkBytes == 0) chunkBytes = 1;
     if (!docStoreResolver_ || !mirrorHealthy_ || !mirrorDriftCount_) return result;
 
     // One resolved row: the project decides which env (and therefore which
@@ -906,9 +957,27 @@ MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
         ++result.failed;
     };
 
+    // ⚠ v2.11.0 final review (finding 3) — Deferred holds NO Document.
+    //
+    // It used to move the whole Row in, Document included. That is fine while
+    // failures are sparse, but MDB_MAP_FULL, ENOSPC or a poisoned handle makes
+    // EVERY put throw: put_batch aborts, every row is then retried alone and
+    // fails alone, and `deferred` accumulated the ENTIRE touched set. Under
+    // --recovery-mode=wal_only on this repo's measured ~2.5-2.9 MB document
+    // shapes that is the tens-of-GB peak the streaming rewrite had just
+    // removed, so a disk-full boot turned from "degraded, rows stale" into an
+    // OOM-killed boot loop - and those are exactly the states where the boot
+    // must still complete.
+    //
+    // Phase 3 re-reads each document from MemoryStore instead, with the same
+    // lookup the resolve step already uses, so at most ONE document is
+    // resident there regardless of how many rows failed. A row that has since
+    // been evicted or deleted simply has nothing left to mirror.
     struct Deferred {
         smartbotic::db::storage::DocumentStore* ds;
-        Row row;
+        std::string qualified;
+        std::string bare;
+        std::string id;
     };
     std::vector<Deferred> deferred;
 
@@ -931,10 +1000,19 @@ MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
     // been tens of GB of document clones resident at once, on the boot path,
     // before READY. A slow boot completes; a killed one does not.
     //
-    // Now at most `chunkSize` documents (plus `deferred`, which is small by
-    // construction - only rows that failed their own transaction) are ever
-    // resident. Each window is resolved, committed, and released before the
-    // next window is resolved.
+    // Each window is resolved, committed, and released before the next window
+    // is resolved.
+    //
+    // ⚠ v2.11.0 final review (finding 3) — the window is bounded by BYTES as
+    // well as by count. `chunkSize` alone is a proxy that this repository's own
+    // measurements falsify: at 256 rows of ~2.9 MB each (anime_images) one
+    // window is ~750 MB of resident document copies, plus as much again in
+    // LMDB dirty pages inside the transaction, on the boot path, in the HAPPY
+    // path. The byte budget is what actually bounds peak memory; the count cap
+    // stays because it also bounds the fsync-per-commit amortisation for small
+    // documents. Whichever limit is hit first closes the window. A single
+    // document larger than the whole budget still forms a window of one, so
+    // progress is guaranteed.
     //
     // Parsing happens per row inside the window, NOT inside the transaction,
     // precisely so a malformed collection key cannot throw from a
@@ -943,13 +1021,17 @@ MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
     // and committed once per project present in that window. Worst case -
     // every row in a different project - degrades to a transaction per row,
     // which is exactly the pre-batching cost and never worse.
-    for (size_t windowStart = 0; windowStart < docs.size(); windowStart += chunkSize) {
-        const size_t windowEnd = std::min(windowStart + chunkSize, docs.size());
-
+    size_t cursor = 0;
+    while (cursor < docs.size()) {
         std::map<std::string, std::vector<Row>> byProject;
-        for (size_t i = windowStart; i < windowEnd; ++i) {
-            const std::string& collection = docs[i].first;
-            const std::string& id = docs[i].second;
+        size_t inWindow = 0;
+        uint64_t windowBytes = 0;
+
+        while (cursor < docs.size() && inWindow < chunkSize && windowBytes < chunkBytes) {
+            const std::string& collection = docs[cursor].first;
+            const std::string& id = docs[cursor].second;
+            ++cursor;
+
             std::string project;
             std::string bare;
             try {
@@ -973,6 +1055,10 @@ MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
                 if (it == coll->documents.end()) continue;  // nothing to mirror
                 doc = it->second;
             }
+            // Counted only for rows that actually entered the window: skipped
+            // rows cost nothing and must not shrink it.
+            ++inWindow;
+            windowBytes += estimateDocumentSize(*doc);
             byProject[project].push_back(Row{collection, std::move(bare), id, std::move(*doc)});
         }
 
@@ -1008,7 +1094,7 @@ MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
                     ds->put(r.bare, r.id, r.doc);
                     ++result.remirrored;
                 } catch (const std::exception&) {
-                    deferred.push_back(Deferred{ds, std::move(r)});
+                    deferred.push_back(Deferred{ds, r.qualified, r.bare, r.id});
                 }
             }
         }
@@ -1022,16 +1108,34 @@ MemoryStore::RemirrorBatchResult MemoryStore::remirrorDocuments(
     // until A itself has been re-mirrored. Exactly one extra attempt - no
     // loop, so a genuinely unresolvable conflict cannot spin the boot path.
     //
-    // ⚠ NOT COVERED BY ANY TEST: no test in the suite induces a
-    // UniqueViolation from this pass, so replacing this retry with a bare
-    // noteFailure() would still pass everything. Recorded in the task-12
-    // report rather than papered over.
+    // ⚠ v2.11.0 final review (finding 4) — this retry is REACHABLE, and only
+    // became so with that finding's fix. While the pass ran inside recover(),
+    // before applyIndexDeclarations() had armed anything, unique_fields_ was
+    // empty for every collection and put() could not raise UniqueViolation at
+    // all, which made this whole phase dead code justified by dead reasoning.
+    // Now that the pass runs after arming, a moved unique value genuinely
+    // conflicts against the other row's not-yet-re-mirrored posting, and this
+    // is what clears it. Pinned by
+    // test_moved_unique_value_is_resolved_by_the_retry_pass, which fails on
+    // five assertions if this retry is replaced by a bare noteFailure().
+    //
+    // The document is RE-READ here rather than carried (finding 3): see the
+    // Deferred comment above.
     for (const auto& d : deferred) {
+        const CollectionData* coll = getCollection(d.qualified);
+        if (!coll) continue;   // collection gone; nothing left to mirror
+        std::optional<Document> doc;
+        {
+            std::shared_lock<std::shared_mutex> lock(coll->mutex);
+            auto it = coll->documents.find(d.id);
+            if (it == coll->documents.end()) continue;
+            doc = it->second;
+        }
         try {
-            d.ds->put(d.row.bare, d.row.id, d.row.doc);
+            d.ds->put(d.bare, d.id, *doc);
             ++result.remirrored;
         } catch (const std::exception& e) {
-            noteFailure(d.row.qualified, d.row.id, e.what());
+            noteFailure(d.qualified, d.id, e.what());
         }
     }
 
@@ -1089,11 +1193,15 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
     // Track memory change (old size)
     uint64_t oldSize = estimateDocumentSize(it->second);
 
-    // v2.11.0 T11 — snapshot for undo, same reasoning as update() above.
-    const Document original = it->second;
+    // v2.11.0 T11 — snapshot for undo, same reasoning as update() above,
+    // including finding 5's gating (see update()).
+    const bool wantUndo = needsUndoSnapshot(collection);
+    const Document original = wantUndo ? it->second : Document{};
     std::optional<std::vector<float>> originalVec;
-    if (auto vit = coll->vectors.find(id); vit != coll->vectors.end()) {
-        originalVec = vit->second;
+    if (wantUndo) {
+        if (auto vit = coll->vectors.find(id); vit != coll->vectors.end()) {
+            originalVec = vit->second;
+        }
     }
 
     // Remove from old expiration index
@@ -1165,6 +1273,20 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
     // v2.11.0 T11 — same undo shape as update(): restore doc/vector/expiry
     // to the pre-write snapshot on a rejected write.
     mirrorDocOrUndo(collection, id, updated, EventType::UPDATE, [&]() {
+        // finding 5 — the snapshot is conditional, so the undo must be too.
+        // With no rejecting constraint declared, `original` is an EMPTY
+        // Document and restoring from it would DESTROY the row, so refuse and
+        // leave the mutation in place (the pre-T11 behaviour, and the lesser
+        // harm). Unreachable in practice: put() cannot raise
+        // UniqueViolation/MissingParentReference without a declaration, which
+        // is the same condition needsUndoSnapshot() tests.
+        if (!wantUndo) {
+            spdlog::error("mirror rejected a write to '{}/{}' but no undo snapshot "
+                          "was taken (no unique field or validate_on_write relation "
+                          "is declared for this collection) - the in-memory row is "
+                          "left as written", collection, id);
+            return;
+        }
         if (updated.expiresAt > 0) {
             removeFromExpirationIndex(*coll, id, updated.expiresAt);
         }
@@ -1227,10 +1349,14 @@ uint64_t MemoryStore::patchDocument(const std::string& collection, const std::st
     // v2.11.0 T11 — snapshot for undo. patchDocument mutates it->second in
     // place rather than building a separate "updated" object, so the undo
     // needs the whole pre-patch document plus whatever vector it held.
-    const Document original = it->second;
+    // Gated per finding 5 - see update().
+    const bool wantUndo = needsUndoSnapshot(collection);
+    const Document original = wantUndo ? it->second : Document{};
     std::optional<std::vector<float>> originalVec;
-    if (auto vit = coll->vectors.find(id); vit != coll->vectors.end()) {
-        originalVec = vit->second;
+    if (wantUndo) {
+        if (auto vit = coll->vectors.find(id); vit != coll->vectors.end()) {
+            originalVec = vit->second;
+        }
     }
 
     // Remove from old expiration index
@@ -1292,6 +1418,18 @@ uint64_t MemoryStore::patchDocument(const std::string& collection, const std::st
     // the patch attempt left behind first). Vector is only touched back if
     // the patch attempt itself touched it.
     mirrorDocOrUndo(collection, id, updatedDoc, EventType::UPDATE, [&]() {
+        // finding 5 — refuse to restore from a snapshot that was never taken
+        // (see the note on the snapshot itself); an empty Document would
+        // destroy the row. Leaving the mutation in place is the pre-T11
+        // behaviour and the lesser harm, and it is unreachable anyway: put()
+        // cannot reject without a declaration.
+        if (!wantUndo) {
+            spdlog::error("mirror rejected a write to '{{}}/{{}}' but no undo snapshot "
+                          "was taken (no unique field or validate_on_write relation "
+                          "is declared for this collection) - the in-memory row is "
+                          "left as written", collection, id);
+            return;
+        }
         if (it->second.expiresAt > 0) {
             removeFromExpirationIndex(*coll, id, it->second.expiresAt);
         }
@@ -1608,7 +1746,9 @@ bool MemoryStore::setAdd(const std::string& collection, const std::string& setId
         }
     }
 
-    const Document original = it->second;
+    // v2.11.0 T11 snapshot for undo, gated per finding 5 - see update().
+    const bool wantUndo = needsUndoSnapshot(collection);
+    const Document original = wantUndo ? it->second : Document{};
 
     saveToHistory(*coll, it->second);
     members.push_back(member);
@@ -1623,6 +1763,18 @@ bool MemoryStore::setAdd(const std::string& collection, const std::string& setId
     // v2.0 dual-write under lock (sadd-style: append member, update doc)
     // v2.11.0 T11 — undo restores the document to its pre-add state.
     mirrorDocOrUndo(collection, setId, updated, EventType::UPDATE, [&]() {
+        // finding 5 — refuse to restore from a snapshot that was never taken
+        // (see the note on the snapshot itself); an empty Document would
+        // destroy the row. Leaving the mutation in place is the pre-T11
+        // behaviour and the lesser harm, and it is unreachable anyway: put()
+        // cannot reject without a declaration.
+        if (!wantUndo) {
+            spdlog::error("mirror rejected a write to '{{}}/{{}}' but no undo snapshot "
+                          "was taken (no unique field or validate_on_write relation "
+                          "is declared for this collection) - the in-memory row is "
+                          "left as written", collection, setId);
+            return;
+        }
         it->second = original;
     });
 
@@ -1675,7 +1827,9 @@ bool MemoryStore::setRemove(const std::string& collection, const std::string& se
     // unique constraint on _version or _updated_at makes to_add non-empty
     // here too, so this path can raise UniqueViolation same as any other
     // UPDATE mirror call, and needs the same undo as setAdd's update branch.
-    const Document original = it->second;
+    // Gated per finding 5 - see update().
+    const bool wantUndo = needsUndoSnapshot(collection);
+    const Document original = wantUndo ? it->second : Document{};
 
     saveToHistory(*coll, it->second);
     tree["members"] = newMembers;
@@ -1689,6 +1843,18 @@ bool MemoryStore::setRemove(const std::string& collection, const std::string& se
 
     // v2.0 dual-write under lock (srem-style: trim members, update doc)
     mirrorDocOrUndo(collection, setId, updated, EventType::UPDATE, [&]() {
+        // finding 5 — refuse to restore from a snapshot that was never taken
+        // (see the note on the snapshot itself); an empty Document would
+        // destroy the row. Leaving the mutation in place is the pre-T11
+        // behaviour and the lesser harm, and it is unreachable anyway: put()
+        // cannot reject without a declaration.
+        if (!wantUndo) {
+            spdlog::error("mirror rejected a write to '{{}}/{{}}' but no undo snapshot "
+                          "was taken (no unique field or validate_on_write relation "
+                          "is declared for this collection) - the in-memory row is "
+                          "left as written", collection, setId);
+            return;
+        }
         it->second = original;
     });
 
@@ -2346,7 +2512,10 @@ uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std:
     Document restoredDoc;
 
     // v2.11.0 T11 — snapshot for undo on the "active document" branch.
-    const Document originalIfActive = wasDeleted ? Document{} : docIt->second;
+    // Gated per finding 5 - see update().
+    const bool wantUndo = needsUndoSnapshot(collection);
+    const Document originalIfActive =
+        (wasDeleted || !wantUndo) ? Document{} : docIt->second;
 
     if (!wasDeleted) {
         // Active document: save current state to history, then overwrite
@@ -2403,6 +2572,16 @@ uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std:
     // back exactly as it was before the restore.
     mirrorDocOrUndo(collection, id, restoredDoc,
                     wasDeleted ? EventType::INSERT : EventType::UPDATE, [&]() {
+        // finding 5 — the wasDeleted branch's undo needs no snapshot (it just
+        // erases what it recreated). The active branch restores from
+        // `originalIfActive` and must refuse if that was never captured.
+        if (!wasDeleted && !wantUndo) {
+            spdlog::error("mirror rejected a version restore of '{}/{}' but no undo "
+                          "snapshot was taken (no unique field or validate_on_write "
+                          "relation is declared for this collection) - the in-memory "
+                          "row is left as written", collection, id);
+            return;
+        }
         if (restoredDoc.expiresAt > 0) {
             removeFromExpirationIndex(*coll, id, restoredDoc.expiresAt);
         }
@@ -2822,6 +3001,22 @@ void MemoryStore::mirrorWriteToDocStore(const std::string& collection, const std
         pc.collection, id, doc, eventType);
 }
 
+bool MemoryStore::needsUndoSnapshot(const std::string& collection) const {
+    // No mirror => no rejection is possible, but also no cost worth reasoning
+    // about; report true so the ONE rule callers follow ("snapshot iff this
+    // says so") never has an exception, and because mirrorDocOrUndo is a no-op
+    // in that state anyway.
+    if (!docStoreResolver_) return true;
+    try {
+        auto pc = smartbotic::database::parseProjectCollection(collection);
+        auto* ds = docStoreResolver_(pc.project);
+        if (!ds) return true;
+        return ds->can_reject_writes(pc.collection);
+    } catch (const std::exception&) {
+        return true;   // fail safe - see the header
+    }
+}
+
 void MemoryStore::mirrorDocOrUndo(const std::string& collection, const std::string& id,
                                    const Document& doc, EventType eventType,
                                    const std::function<void()>& undo) {

service/src/memory_store.hpp

@@ -194,6 +194,54 @@ public:
                                 std::atomic<bool>* healthy,
                                 std::atomic<uint64_t>* drift);
 
+    /**
+     * v2.11.0 final review (finding 6) — read-only view of the mirror's
+     * health, for code that holds a MemoryStore& but not a DatabaseService&
+     * (relations/relation_cascade.cpp). DatabaseService exposes the same two
+     * values to the gRPC layer; these read the very same atomics.
+     *
+     * ⚠ "No mirror wired" reports HEALTHY with zero drift, deliberately: with
+     * no document store there is no second substrate that could be behind,
+     * so there is nothing to be unhealthy about. This matches
+     * mirrorWriteToDocStore(), which silently no-ops in the same state rather
+     * than treating it as a fault, and it keeps in-process unit fixtures that
+     * never wire a mirror from being refused every destructive operation.
+     */
+    [[nodiscard]] bool mirrorHealthy() const noexcept {
+        return mirrorHealthy_ == nullptr || mirrorHealthy_->load(std::memory_order_relaxed);
+    }
+    [[nodiscard]] uint64_t mirrorDriftCount() const noexcept {
+        return mirrorDriftCount_ == nullptr
+            ? 0
+            : mirrorDriftCount_->load(std::memory_order_relaxed);
+    }
+
+    /**
+     * v2.11.0 final review (finding 5) — does a write to `collection` need
+     * T11's pre-write undo snapshot?
+     *
+     * True only when the mirror can REJECT the write (a declared unique field,
+     * or a relation with validateOnWrite) - which is the only situation where
+     * mirrorDocOrUndo()'s undo can ever run. T11 added
+     * `const Document original = it->second;` to update, updateIfVersion,
+     * upsert, patchDocument, setAdd, setRemove and restoreToVersion
+     * unconditionally. Document's copy constructor is
+     * yyjson_mut_val_mut_copy: O(size), a fresh yyjson_mut_doc, taken inside
+     * the collection's unique_lock. On this repo's measured document shapes
+     * that is millions of bytes and milliseconds of allocation per write, on
+     * every install, added to the path v2.8.0 spent a release making cheaper -
+     * and where nothing is declared, the copy is never read.
+     *
+     * Fails SAFE: no mirror wired, an unresolvable collection name, or a
+     * backend that does not implement the query all answer true, so the
+     * snapshot is taken. A wrong `true` costs a copy; a wrong `false` would
+     * leave a rejected write's mutation sitting in MemoryStore unenforced,
+     * which is why the guard inside each undo lambda refuses to restore from a
+     * snapshot that was never taken rather than clobbering the row with an
+     * empty Document.
+     */
+    [[nodiscard]] bool needsUndoSnapshot(const std::string& collection) const;
+
     // ===== Collection Management =====
 
     /**
@@ -321,10 +369,37 @@ public:
      * loadDocumentWithHistory() do not, and LMDB then permanently serves
      * the pre-cascade row until something else happens to rewrite it.
      *
-     * PersistenceManager::recover() re-mirrors once per DISTINCT id that
-     * WAL replay applied as INSERT/UPDATE/UPSERT (deduplicated, and skipped
-     * if a later DELETE for that id was also replayed — remove() already
-     * mirrored that) — a targeted post-replay pass, not a mirror call on
+     * ⚠ WHEN THIS RUNS, and why it matters (v2.11.0 final review, finding 4).
+     * PersistenceManager::recover() only COLLECTS the id list, into
+     * RecoveryOutcome::pendingRemirror. The pass itself runs from
+     * DatabaseService::initialize(), AFTER applyRelationDeclarations() and
+     * applyIndexDeclarations(), via PersistenceManager::runPendingRemirror().
+     *
+     * It used to run inside recover(), which is line 88 of initialize() while
+     * the declarations are armed at lines ~199 and ~208 - so put() read
+     * indexed_fields(collection), unique_fields(collection) and
+     * relations(collection) while ALL THREE MAPS WERE STILL EMPTY. The pass
+     * therefore repaired the document and corrupted everything derived from
+     * it:
+     *   - no secondary-index maintenance, so the document moved forward in
+     *     LMDB while its postings did not. An index-served query then returns
+     *     the row under its OLD value and misses it under its new one -
+     *     silently wrong rows, precisely what putting maintainIndexes() inside
+     *     the write transaction exists to prevent. This bit ANY install with a
+     *     v2.9 index declared, relations or not.
+     *   - no reverse-index maintenance, so relation_cascade.hpp's claim that
+     *     this pass closes "a live reverse-index posting pointing at a
+     *     now-deleted parent" was false.
+     *   - UniqueViolation was unreachable, which made phase 3's retry and the
+     *     insertion-ORDERED tracking in persistence_manager.cpp dead code
+     *     justified by dead reasoning.
+     * backfillIntoDocStore() already sits at that point in initialize() and is
+     * already guarded for these exceptions, so the ordering is precedented.
+     *
+     * One entry per DISTINCT id that WAL replay applied as
+     * INSERT/UPDATE/UPSERT (deduplicated, and skipped if a later DELETE for
+     * that id was also replayed — remove() already mirrored that) — a
+     * targeted post-replay pass, not a mirror call on
      * every replayed entry, which would multiply LMDB writes by however
      * many times a hot document was rewritten since the last snapshot for
      * no benefit (the ordinary case's LMDB write already happened at
@@ -417,7 +492,8 @@ public:
      */
     RemirrorBatchResult remirrorDocuments(
         const std::vector<std::pair<std::string, std::string>>& docs,
-        size_t chunkSize = 256);
+        size_t chunkSize = 256,
+        uint64_t chunkBytes = 64ull * 1024 * 1024);
 
     /**
      * Check if a document exists.

service/src/persistence/persistence_manager.cpp

@@ -231,6 +231,14 @@ RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
         // transient unique-index conflict, and an unordered_map made that
         // non-deterministic between boots. WAL order is the order the writes
         // originally happened in, which is the least surprising choice.
+        //
+        // v2.11.0 final review (finding 4) — this reasoning is LIVE again. It
+        // was dead while the pass ran inside recover(): unique_fields_ was
+        // still empty at that point, so no transient conflict could occur and
+        // the ordering decided nothing. The pass now runs from
+        // DatabaseService::initialize() after applyIndexDeclarations(), so a
+        // unique index really is armed and the order really does pick a
+        // winner.
         std::vector<std::pair<std::string, std::string>> touched;
         std::unordered_map<std::string, size_t> touchedIndex;
         std::vector<bool> touchedAlive;
@@ -260,28 +268,19 @@ RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
         for (size_t i = 0; i < touched.size(); ++i) {
             if (touchedAlive[i]) toRemirror.push_back(touched[i]);
         }
-        // BATCHED, and it never throws — both deliberate, see
-        // MemoryStore::remirrorDocuments(). Per-document transactions here
-        // meant one fsync per replayed id on the boot path before
-        // sd_notify(READY=1); an escaping exception here meant recover()
-        // failed and the service refused to start, permanently, on data that
-        // booted fine before.
-        auto remirror = store.remirrorDocuments(toRemirror);
-        outcome.updatesRemirroredAfterReplay = remirror.remirrored;
-        outcome.updatesRemirrorFailed = remirror.failed;
-        if (remirror.remirrored > 0) {
-            spdlog::info("Re-mirrored {} document(s) to LMDB after WAL replay "
-                        "(closes the window where a WAL-first writer, e.g. a "
-                        "cascade delete, logged an UPDATE whose LMDB write "
-                        "never ran before a crash)", remirror.remirrored);
-        }
-        if (remirror.failed > 0) {
-            spdlog::error("{} document(s) could not be re-mirrored to LMDB after WAL "
-                          "replay (each logged above with its collection and id). "
-                          "Those rows stay stale in LMDB; recovery is NOT failed by "
-                          "this - a repair pass that cannot repair one row must not "
-                          "stop the service from starting.", remirror.failed);
-        }
+        // v2.11.0 final review (finding 4) — HANDED OFF, not run here.
+        // recover() runs at DatabaseService::initialize() line ~88, while
+        // relation and index declarations are armed at ~199/~208. Running the
+        // pass here meant LmdbDocumentStore::put() consulted
+        // indexed_fields()/unique_fields()/relations() while all three were
+        // still empty: the document was repaired and every index posting
+        // derived from it was left stale, which an index-served query then
+        // returns as a silently wrong row. initialize() calls
+        // runPendingRemirror() after arming instead.
+        outcome.pendingRemirror = std::move(toRemirror);
+        spdlog::info("Collected {} document(s) for the post-replay re-mirror pass "
+                     "(runs after index/relation declarations are armed)",
+                     outcome.pendingRemirror.size());
         // Anchor the WAL's sequence_ counter to the snapshot's walSequence.
         // Without this, the steady-state outcome of `truncateBefore(walSeq)`
         // after every snapshot — which deletes every WAL file because all
@@ -444,6 +443,32 @@ RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
     return outcome;
 }
 
+void PersistenceManager::runPendingRemirror(MemoryStore& store, RecoveryOutcome& outcome) {
+    if (outcome.pendingRemirror.empty()) return;
+
+    // BATCHED, and it never throws — both deliberate, see
+    // MemoryStore::remirrorDocuments(). Per-document transactions meant one
+    // fsync per replayed id on the boot path before sd_notify(READY=1); an
+    // escaping exception meant the service refused to start, permanently, on
+    // data that booted fine before.
+    auto remirror = store.remirrorDocuments(outcome.pendingRemirror);
+    outcome.updatesRemirroredAfterReplay = remirror.remirrored;
+    outcome.updatesRemirrorFailed = remirror.failed;
+    if (remirror.remirrored > 0) {
+        spdlog::info("Re-mirrored {} document(s) to LMDB after WAL replay "
+                    "(closes the window where a WAL-first writer, e.g. a "
+                    "cascade delete, logged an UPDATE whose LMDB write "
+                    "never ran before a crash)", remirror.remirrored);
+    }
+    if (remirror.failed > 0) {
+        spdlog::error("{} document(s) could not be re-mirrored to LMDB after WAL "
+                      "replay (each logged above with its collection and id). "
+                      "Those rows stay stale in LMDB; recovery is NOT failed by "
+                      "this - a repair pass that cannot repair one row must not "
+                      "stop the service from starting.", remirror.failed);
+    }
+}
+
 uint64_t PersistenceManager::logInsert(const std::string& collection, const Document& doc,
                                         const std::string& originNodeId) {
     if (!running_.load()) return 0;

service/src/persistence/persistence_manager.hpp

@@ -12,6 +12,8 @@
 #include <mutex>
 #include <thread>
 #include <unordered_map>
+#include <utility>
+#include <vector>
 
 namespace smartbotic::database {
 
@@ -72,6 +74,22 @@ struct RecoveryOutcome {
     // from starting, which is what an unguarded throw here used to do.
     uint64_t updatesRemirrorFailed = 0;
 
+    // v2.11.0 final review (finding 4) — the (collection, id) pairs WAL
+    // replay applied as INSERT/UPDATE/UPSERT and that still need mirroring
+    // to LMDB. recover() only COLLECTS this list; the pass itself runs from
+    // DatabaseService::initialize() via runPendingRemirror(), AFTER
+    // applyRelationDeclarations()/applyIndexDeclarations().
+    //
+    // ⚠ THE ORDERING IS THE POINT, not a refactor. Run from inside
+    // recover(), the pass wrote documents while indexed_fields_,
+    // unique_fields_ and relations_ were all still empty, so it repaired the
+    // document and left every index and reverse-index posting derived from
+    // it stale - silently wrong rows on any install with a v2.9 index
+    // declared. See MemoryStore::remirrorDocuments()'s doc comment.
+    //
+    // Empty after a recovery that replayed nothing, and after ForceEmpty.
+    std::vector<std::pair<std::string, std::string>> pendingRemirror;
+
     bool isNonTrivial() const {
         return kind != Kind::TrivialSuccess && kind != Kind::FreshInstall;
     }
@@ -142,6 +160,24 @@ public:
      */
     RecoveryOutcome recover(MemoryStore& store);
 
+    /**
+     * v2.11.0 final review (finding 4) — run the post-replay re-mirror pass
+     * over outcome.pendingRemirror and record its counts back onto `outcome`.
+     *
+     * MUST be called AFTER the caller has armed relation and index
+     * declarations (DatabaseService::initialize() calls it right after
+     * applyIndexDeclarations()), because LmdbDocumentStore::put() reads
+     * indexed_fields()/unique_fields()/relations() live and maintains every
+     * index inside the document's own write transaction. Running it earlier -
+     * which is what recover() itself used to do - repairs the document and
+     * leaves every posting derived from it stale.
+     *
+     * Idempotent, and never throws (MemoryStore::remirrorDocuments guarantees
+     * that; a repair pass must not be able to stop the service from
+     * starting). Safe to call with an empty list.
+     */
+    void runPendingRemirror(MemoryStore& store, RecoveryOutcome& outcome);
+
     /**
      * Log an insert operation.
      * `originNodeId` (v1.8.0) is the node that originally produced this

service/src/relations/relation_cascade.cpp

@@ -334,6 +334,39 @@ bool executeCascade(RelationManager& relations,
                     const std::string& qualifiedParentCollection,
                     const std::string& parentId,
                     const CascadeNotifyFn& notify) {
+    // v2.11.0 final review (finding 6) — REFUSE THE DESTRUCTIVE PATH while
+    // the mirror is unhealthy or has drifted.
+    //
+    // planCascade() builds every `updatedDoc` from the LMDB copy
+    // (store.get(...)), and applyCascadeToMemory() then pushes that copy into
+    // MemoryStore via loadDocumentWithHistory(). In the unhealthy/drifted
+    // state LMDB may be BEHIND - that is precisely why reads fall back to
+    // MemoryStore and why six read gates in database_grpc_impl.cpp test this
+    // pair - so the cascade would overwrite MemoryStore's fresher child body
+    // with a stale one. That is real data loss, produced by a repair-shaped
+    // operation, on a state this codebase treats as routine.
+    //
+    // Thrown as CascadeBlocked so the Delete handler maps it to
+    // FAILED_PRECONDITION with the message intact: nothing has been written
+    // anywhere at this point (this is before writeCascadeWal()), which is
+    // exactly the contract CascadeBlocked already carries. Same refusal
+    // shape, and the same reasoning, as the CreateIndex(unique) and
+    // CreateRelation gates.
+    if (!memStore.mirrorHealthy() || memStore.mirrorDriftCount() != 0) {
+        throw CascadeBlocked(
+            "refusing the cascade/set_null delete of '" + qualifiedParentCollection +
+            "/" + parentId + "': the LMDB mirror is " +
+            (memStore.mirrorHealthy() ? "drifted (" + std::to_string(memStore.mirrorDriftCount()) +
+                                         " row(s) known stale)"
+                                      : "unhealthy") +
+            ", so the child documents this cascade would rewrite are read from a substrate "
+            "that may be behind MemoryStore - applying them would overwrite fresher data "
+            "with older data. Restrict/no_action deletes are unaffected. Fix the mirror "
+            "(see the ERROR lines naming the failed collection and id) and restart to "
+            "clear drift, or use on_delete=no_action if the dangling reference is "
+            "acceptable.");
+    }
+
     const CascadePlan plan = planCascade(relations, store, configManager,
                                          qualifiedParentCollection, parentId);
     writeCascadeWal(persistence, qualifiedParentCollection, parentId, plan);

service/src/relations/relation_cascade.hpp

@@ -53,14 +53,28 @@
 // now-deleted parent, invisible until that child is next rewritten through
 // the ordinary write path (which re-mirrors it).
 //
-// CLOSED (round 3, refined in round 4) by a post-replay re-mirror pass in
-// PersistenceManager::recover(): every DISTINCT (collection, id) that replay
-// applied as INSERT/UPDATE/UPSERT, minus any a later DELETE removed, is
-// pushed to LMDB via MemoryStore::remirrorDocuments() once replay finishes.
+// CLOSED (round 3, refined in round 4) by a post-replay re-mirror pass:
+// every DISTINCT (collection, id) that replay applied as
+// INSERT/UPDATE/UPSERT, minus any a later DELETE removed, is collected by
+// PersistenceManager::recover() into RecoveryOutcome::pendingRemirror and
+// pushed to LMDB via MemoryStore::remirrorDocuments().
 // It is BATCHED (chunked transactions — one fsync per document on the boot
 // path was minutes of startup on a large WAL) and it NEVER THROWS (an
 // escaping exception from a repair pass turned recover() into a refusal to
-// start). It is deliberately not cascade-specific: WAL entries carry no
+// start).
+//
+// ⚠ THE REVERSE-INDEX HALF OF THIS ONLY BECAME TRUE IN v2.11.0's FINAL
+// REVIEW (finding 4). While the pass ran inside recover() it wrote documents
+// before DatabaseService::initialize() had called
+// applyRelationDeclarations(), so LmdbDocumentStore::relations(collection)
+// was empty and maintainRelations() did nothing: the child document
+// converged and the stale posting named above did NOT. Same for secondary
+// indexes and for unique constraints. The pass now runs from initialize()
+// after both arming steps (PersistenceManager::runPendingRemirror()), which
+// is what makes "plus a live reverse-index posting pointing at a now-deleted
+// parent" actually closed rather than merely claimed. Do not move it back.
+//
+// It is deliberately not cascade-specific: WAL entries carry no
 // "came from a cascade" marker, and the same divergence occurs whenever
 // applyDualWriteMirror() SWALLOWED an ordinary write's LMDB fault (it
 // swallows everything but UniqueViolation), so a WAL entry never did imply
@@ -117,7 +131,22 @@
 // what is likely a rare schema shape, but worth an operator knowing why
 // their delete refuses.
 //
+// ⚠ REFUSED WHILE THE MIRROR IS UNHEALTHY OR DRIFTED (v2.11.0 final review,
+// finding 6). executeCascade() checks MemoryStore::mirrorHealthy() and
+// mirrorDriftCount() as its FIRST action and throws CascadeBlocked if either
+// says the two substrates have diverged. planCascade() builds every
+// `updatedDoc` from the LMDB copy and applyCascadeToMemory() then pushes that
+// copy into MemoryStore, so in a state where LMDB is legitimately BEHIND -
+// which is the entire reason reads fall back to MemoryStore, and why six read
+// gates in database_grpc_impl.cpp test this same pair - the cascade would
+// overwrite MemoryStore's fresher child body with a stale one. That is real
+// data loss caused by a repair-shaped operation. Nothing is written before
+// the check, so the refusal is a clean no-op, and restrict/no_action deletes
+// are unaffected. Same reasoning and same shape as the CreateIndex(unique)
+// and CreateRelation gates.
+//
 // Sequence, in full (mirrors the design doc):
+//   0. Refuse if the LMDB mirror is unhealthy or drifted (see above).
 //   1. Resolve children through the reverse index — planCascade(). Also
 //      where the restrict/grandchild check above runs, and where
 //      CascadeBlocked can be thrown.

service/src/relations/relation_manager.cpp

@@ -16,23 +16,6 @@ uint64_t nowMs() {
         std::chrono::system_clock::now().time_since_epoch()).count();
 }
 
-std::string onDeleteToString(OnDelete v) {
-    switch (v) {
-        case OnDelete::Restrict: return "restrict";
-        case OnDelete::Cascade: return "cascade";
-        case OnDelete::SetNull: return "set_null";
-        case OnDelete::NoAction: return "no_action";
-    }
-    return "restrict";
-}
-
-OnDelete onDeleteFromString(const std::string& s) {
-    if (s == "cascade") return OnDelete::Cascade;
-    if (s == "set_null") return OnDelete::SetNull;
-    if (s == "no_action") return OnDelete::NoAction;
-    return OnDelete::Restrict;
-}
-
 nlohmann::json toJson(const RelationInfo& r) {
     return nlohmann::json{
         {"name", r.name},
@@ -52,7 +35,26 @@ RelationInfo fromJson(const nlohmann::json& j) {
     r.child = j.value("child", "");
     r.childField = j.value("child_field", "");
     r.parent = j.value("parent", "");
-    r.onDelete = onDeleteFromString(j.value("on_delete", "restrict"));
+    // v2.11.0 final review (finding 9) — the persisted value goes through the
+    // same single parser as the RPC. A record whose on_delete is not one of
+    // the four falls back to Restrict, which is the safe direction (it
+    // refuses deletes rather than performing an unintended destructive one),
+    // but it is logged at ERROR: silently coercing is what made a typo
+    // indistinguishable from a deliberate choice. Deliberately NOT skipped
+    // the way a cross-project record is: dropping the relation entirely would
+    // remove protection, which is strictly worse than over-restricting.
+    {
+        const std::string raw = j.value("on_delete", std::string("restrict"));
+        if (auto parsed = parseOnDelete(raw)) {
+            r.onDelete = *parsed;
+        } else {
+            r.onDelete = OnDelete::Restrict;
+            spdlog::error("RelationManager: relation '{}' has an unrecognised "
+                          "on_delete '{}' (expected {}); treating it as 'restrict' "
+                          "- fix the declaration, this is not what was asked for",
+                          r.name, raw, kOnDeleteValues);
+        }
+    }
     r.validateOnWrite = j.value("validate_on_write", false);
     r.createdAt = j.value("created_at", uint64_t{0});
     r.updatedAt = j.value("updated_at", uint64_t{0});
@@ -61,6 +63,33 @@ RelationInfo fromJson(const nlohmann::json& j) {
 
 } // anonymous namespace
 
+std::string onDeleteToString(OnDelete v) {
+    switch (v) {
+        case OnDelete::Restrict: return "restrict";
+        case OnDelete::Cascade: return "cascade";
+        case OnDelete::SetNull: return "set_null";
+        case OnDelete::NoAction: return "no_action";
+    }
+    return "restrict";
+}
+
+std::optional<OnDelete> parseOnDelete(const std::string& s) {
+    if (s == "restrict") return OnDelete::Restrict;
+    if (s == "cascade") return OnDelete::Cascade;
+    if (s == "set_null") return OnDelete::SetNull;
+    if (s == "no_action") return OnDelete::NoAction;
+    return std::nullopt;
+}
+
+std::string RelationManager::canonical(const std::string& name) {
+    if (name.empty()) return name;
+    try {
+        return resolveCollection(name).qualified;
+    } catch (const std::exception&) {
+        return name;   // see the header: unparseable matches nothing, never throws
+    }
+}
+
 RelationManager::RelationManager(MemoryStore& store) : store_(store) {}
 
 void RelationManager::loadFromStore() {
@@ -76,6 +105,9 @@ void RelationManager::loadFromStore() {
     // as a bug in ViewManager, PolicyManager and CollectionConfigManager.
     constexpr uint32_t kPage = 500;
     uint32_t offset = 0;
+    // Collected during the walk and applied after it: mutating _relations
+    // while paging over it would shift the offsets underneath us.
+    std::vector<std::pair<std::string, RelationInfo>> rekeys;
     while (true) {
         Query q;
         q.limit = kPage;
@@ -104,6 +136,15 @@ void RelationManager::loadFromStore() {
                 const auto rn = resolveCollection(r.name);
                 const auto rc = resolveCollection(r.child);
                 const auto rp = resolveCollection(r.parent);
+                // v2.11.0 final review (finding 8) — canonicalise on load, so
+                // the cache is keyed the same way regardless of what form the
+                // stored record used. Legacy/hand-written records naming
+                // "exec_wf" and "default:exec_wf" now resolve to the same
+                // entry instead of coexisting as two relations that arm the
+                // same collection and overwrite each other.
+                r.name = rn.qualified;
+                r.child = rc.qualified;
+                r.parent = rp.qualified;
                 if (rc.project != rp.project || rc.project != rn.project) {
                     spdlog::error(
                         "RelationManager: skipping relation '{}' - name/child/parent "
@@ -119,11 +160,38 @@ void RelationManager::loadFromStore() {
                 continue;
             }
 
+            // v2.11.0 final review (finding 8) — a record stored under a
+            // non-canonical document id is re-keyed in the store as well,
+            // not just in the cache: dropRelation() removes by the canonical
+            // name, so leaving the row under its old id would make the
+            // relation undroppable. Same migration shape ViewManager uses for
+            // its legacy bare-named views. Advisory - a failed re-key leaves
+            // the cache correct and logs, it does not fail the load.
+            if (d.id != r.name) {
+                rekeys.push_back({d.id, r});
+            }
+
             cache_[r.name] = std::move(r);
         }
         if (res.documents.size() < kPage) break;
         offset += kPage;
     }
+
+    for (const auto& [oldId, rel] : rekeys) {
+        try {
+            Document rekeyed;
+            rekeyed.id = rel.name;
+            rekeyed.set_data(toJson(rel));
+            store_.upsert(SYSTEM_COLLECTION, rekeyed);
+            store_.remove(SYSTEM_COLLECTION, oldId);
+            spdlog::warn("RelationManager: re-keyed legacy relation '{}' -> '{}'",
+                         oldId, rel.name);
+        } catch (const std::exception& e) {
+            spdlog::warn("RelationManager: could not re-key legacy relation '{}': {}",
+                         oldId, e.what());
+        }
+    }
+
     spdlog::info("RelationManager: loaded {} relation(s) from {}", cache_.size(), SYSTEM_COLLECTION);
 }
 
@@ -162,15 +230,26 @@ bool RelationManager::createRelation(const RelationInfo& r, std::string& errorOu
         return false;
     }
 
+    RelationInfo out = r;
+    // v2.11.0 final review (finding 8) — CANONICALISE BEFORE ANYTHING ELSE.
+    // The declaration is persisted and cached under the canonical form only,
+    // so a caller sending "exec_wf" and one sending "default:exec_wf" now
+    // create, find and collide with the SAME relation. Before this, the raw
+    // string was stored verbatim, which is how a raw-gRPC caller could arm a
+    // second, near-identical relation over the same child collection and
+    // clobber the first one's armed state.
+    out.name = rn.qualified;
+    out.child = rc.qualified;
+    out.parent = rp.qualified;
+
     {
         std::shared_lock<std::shared_mutex> rlock(mutex_);
-        if (cache_.contains(r.name)) {
-            errorOut = "relation '" + r.name + "' already exists";
+        if (cache_.contains(out.name)) {
+            errorOut = "relation '" + out.name + "' already exists";
             return false;
         }
     }
 
-    RelationInfo out = r;
     out.createdAt = nowMs();
     out.updatedAt = out.createdAt;
 
@@ -198,15 +277,19 @@ bool RelationManager::createRelation(const RelationInfo& r, std::string& errorOu
 }
 
 bool RelationManager::dropRelation(const std::string& qualifiedName, std::string& errorOut) {
+    // finding 8 — canonicalise, so dropping "exec_wf" drops the relation a
+    // caller created as "default:exec_wf". Reported back under the name the
+    // CALLER used, since that is the string they can act on.
+    const std::string key = canonical(qualifiedName);
     {
         std::shared_lock<std::shared_mutex> rlock(mutex_);
-        if (!cache_.contains(qualifiedName)) {
+        if (!cache_.contains(key)) {
             errorOut = "relation '" + qualifiedName + "' does not exist";
             return false;
         }
     }
 
-    bool removed = store_.remove(SYSTEM_COLLECTION, qualifiedName);
+    bool removed = store_.remove(SYSTEM_COLLECTION, key);
     if (!removed) {
         errorOut = "failed to remove relation from store";
         return false;
@@ -214,15 +297,16 @@ bool RelationManager::dropRelation(const std::string& qualifiedName, std::string
 
     {
         std::unique_lock<std::shared_mutex> wlock(mutex_);
-        cache_.erase(qualifiedName);
+        cache_.erase(key);
     }
-    spdlog::info("RelationManager: dropped relation '{}'", qualifiedName);
+    spdlog::info("RelationManager: dropped relation '{}'", key);
     return true;
 }
 
 std::optional<RelationInfo> RelationManager::getRelation(const std::string& qualifiedName) const {
+    const std::string key = canonical(qualifiedName);
     std::shared_lock<std::shared_mutex> lock(mutex_);
-    auto it = cache_.find(qualifiedName);
+    auto it = cache_.find(key);
     if (it == cache_.end()) return std::nullopt;
     return it->second;
 }
@@ -246,20 +330,26 @@ std::vector<RelationInfo> RelationManager::listRelations(const std::string& proj
     return out;
 }
 
-std::vector<RelationInfo> RelationManager::relationsWithParent(const std::string& qualifiedCollection) const {
+std::vector<RelationInfo> RelationManager::relationsWithParent(const std::string& collection) const {
+    // finding 8 — canonicalise the needle. Cached `parent` values are always
+    // canonical (createRelation and loadFromStore both make them so), so a
+    // bare argument would otherwise match nothing and silently report "this
+    // collection is nobody's parent", permitting every delete.
+    const std::string key = canonical(collection);
     std::shared_lock<std::shared_mutex> lock(mutex_);
     std::vector<RelationInfo> out;
     for (const auto& [_, r] : cache_) {
-        if (r.parent == qualifiedCollection) out.push_back(r);
+        if (r.parent == key) out.push_back(r);
     }
     return out;
 }
 
-std::vector<RelationInfo> RelationManager::relationsWithChild(const std::string& qualifiedCollection) const {
+std::vector<RelationInfo> RelationManager::relationsWithChild(const std::string& collection) const {
+    const std::string key = canonical(collection);
     std::shared_lock<std::shared_mutex> lock(mutex_);
     std::vector<RelationInfo> out;
     for (const auto& [_, r] : cache_) {
-        if (r.child == qualifiedCollection) out.push_back(r);
+        if (r.child == key) out.push_back(r);
     }
     return out;
 }

service/src/relations/relation_manager.hpp

@@ -17,6 +17,29 @@ class MemoryStore;
  */
 enum class OnDelete { Restrict, Cascade, SetNull, NoAction };
 
+/**
+ * v2.11.0 final review (finding 9) — the SINGLE string->OnDelete parser.
+ *
+ * Returns nullopt for anything that is not exactly one of "restrict",
+ * "cascade", "set_null", "no_action". There used to be two copies of this
+ * (relationOnDeleteFromString in database_grpc_impl.cpp and
+ * onDeleteFromString in relation_manager.cpp) and BOTH silently coerced an
+ * unrecognised string to Restrict. That failed safe while cascade/set_null
+ * were inert, but T12 made those strings destructive in the other
+ * direction: an operator who types "Cascade" was told the relation was
+ * created and believed cascade was armed while `restrict` actually was, so
+ * the deletes they expected to cascade started failing with
+ * FAILED_PRECONDITION instead - and nothing anywhere named the typo.
+ *
+ * Rendering back to a string lives in onDeleteToString(), also here, so the
+ * two directions cannot drift.
+ */
+std::optional<OnDelete> parseOnDelete(const std::string& s);
+std::string onDeleteToString(OnDelete v);
+
+// Every string an on_delete value may take, for error messages.
+inline constexpr const char* kOnDeleteValues = "restrict|cascade|set_null|no_action";
+
 /**
  * In-memory representation of a relation declaration.
  * Mirrors the (future) RelationDefinition proto message.
@@ -90,15 +113,45 @@ public:
      */
     std::vector<RelationInfo> listRelations(const std::string& project = "") const;
 
-    // Relations whose PARENT is this (qualified) collection - what a
-    // delete must consult.
-    std::vector<RelationInfo> relationsWithParent(const std::string& qualifiedCollection) const;
+    // Relations whose PARENT is this collection - what a delete must
+    // consult. See the canonicalisation note below: the argument may be
+    // bare or qualified.
+    std::vector<RelationInfo> relationsWithParent(const std::string& collection) const;
 
-    // Relations whose CHILD is this (qualified) collection - what a
-    // write must maintain.
-    std::vector<RelationInfo> relationsWithChild(const std::string& qualifiedCollection) const;
+    // Relations whose CHILD is this collection - what a write must
+    // maintain. Same canonicalisation as relationsWithParent.
+    std::vector<RelationInfo> relationsWithChild(const std::string& collection) const;
 
 private:
+    /**
+     * v2.11.0 final review (finding 8) — EVERY name crossing this class's
+     * boundary is canonicalised to "<project>:<name>" here, on the way in
+     * and on the way out, and the cache is keyed by the canonical form only.
+     *
+     * This is the STRUCTURAL fix for the bare-vs-qualified bug class, which
+     * this repository has now shipped three times: v2.4.2 lost every view
+     * for two releases (createView qualified `collection` but not `name`),
+     * v2.4.5 gave every collection a phantom twin (createCollection sent a
+     * bare name while insert sent a qualified one), and on this branch
+     * armRelationsForChild looked relations up under the caller's raw
+     * string while boot arming used the canonical one - so a raw-gRPC caller
+     * naming "executions" instead of "default:executions" could REPLACE or
+     * ERASE the real relation's armed state for the life of the process,
+     * logged only as "re-armed N relation(s)", and a restart silently
+     * repaired it.
+     *
+     * Spot-fixing each call site is what produced three occurrences. Doing
+     * it at the one funnel every entry point already goes through (the RPC
+     * handlers, the boot re-arm, the CLI and loadFromStore all reach
+     * relations exclusively through this class) makes the mismatch
+     * unrepresentable instead of merely absent today.
+     *
+     * Unparseable input is returned UNCHANGED rather than throwing: a bad
+     * name then simply matches nothing, which is what the callers already
+     * handle, and no lookup should be able to abort a delete or a boot.
+     */
+    static std::string canonical(const std::string& name);
+
     MemoryStore& store_;
     mutable std::shared_mutex mutex_;
     std::unordered_map<std::string, RelationInfo> cache_;

service/src/storage/document_store.hpp

@@ -108,6 +108,25 @@ public:
     // Drop a collection sub-db entirely. Returns true if it existed.
     virtual bool drop_collection(std::string_view collection) = 0;
 
+    // v2.11.0 final review (finding 5) — can a put() on this collection
+    // REJECT the write (UniqueViolation / MissingParentReference)?
+    //
+    // MemoryStore's update-shaped write paths must snapshot the pre-write
+    // Document so they can undo their in-memory mutation when the mirror
+    // rejects it (see MemoryStore::mirrorDocOrUndo). That copy is
+    // yyjson_mut_val_mut_copy - O(document size), a fresh allocation, taken
+    // inside the collection's unique_lock - and T11 added it unconditionally,
+    // so every install paid it on every update whether or not any unique
+    // field or relation existed. Where none exists the undo can never fire, so
+    // the copy is pure cost, on the exact path v2.8.0 spent a release making
+    // cheaper.
+    //
+    // The DEFAULT IS true - fail safe. A backend that does not know must be
+    // treated as if it can reject, because the alternative (skipping the
+    // snapshot when a rejection is in fact possible) means a rejected write
+    // leaves its mutation in MemoryStore unenforced.
+    virtual bool can_reject_writes(std::string_view /*collection*/) { return true; }
+
     // ==========================================================================
     // v2.0 Stage 5 — vector sub-db operations.
     //

service/src/storage/document_store_lmdb.cpp

@@ -746,6 +746,44 @@ void LmdbDocumentStore::set_relations(std::string_view collection,
     } else {
         relations_[std::string(collection)] = std::move(rels);
     }
+    // finding 5 — count only collections that can actually REJECT a write. A
+    // relation without validateOnWrite still maintains its reverse index on
+    // every put, but it never throws, so it cannot make an undo snapshot
+    // necessary.
+    size_t rejecting = 0;
+    for (const auto& [_, refs] : relations_) {
+        for (const auto& ref : refs) {
+            if (ref.validateOnWrite) { ++rejecting; break; }
+        }
+    }
+    rejecting_relation_collections_.store(rejecting, std::memory_order_relaxed);
+}
+
+bool LmdbDocumentStore::can_reject_writes(std::string_view collection) {
+    // The overwhelmingly common case: nothing anywhere in this process has a
+    // rejecting constraint declared, so no mutex is taken and no string is
+    // constructed. This is the check that keeps T11's undo snapshot off the
+    // write path of every install that does not use unique fields or
+    // validate_on_write.
+    if (rejecting_unique_collections_.load(std::memory_order_relaxed) == 0 &&
+        rejecting_relation_collections_.load(std::memory_order_relaxed) == 0) {
+        return false;
+    }
+    const std::string key(collection);
+    {
+        std::lock_guard<std::mutex> lock(index_mutex_);
+        if (unique_fields_.contains(key)) return true;
+    }
+    {
+        std::lock_guard<std::mutex> lock(relations_mutex_);
+        auto it = relations_.find(key);
+        if (it != relations_.end()) {
+            for (const auto& ref : it->second) {
+                if (ref.validateOnWrite) return true;
+            }
+        }
+    }
+    return false;
 }
 
 std::vector<RelationRef> LmdbDocumentStore::relations(std::string_view collection) {
@@ -1678,6 +1716,12 @@ void LmdbDocumentStore::set_unique_fields(std::string_view collection,
     } else {
         unique_fields_[std::string(collection)] = std::move(fields);
     }
+    // finding 5 — recomputed under the same mutex that owns the map, so the
+    // counter can never disagree with it. Recomputed rather than incremented:
+    // this setter is idempotent-by-replacement, so a blind ++ would double
+    // count a re-declaration.
+    rejecting_unique_collections_.store(unique_fields_.size(),
+                                       std::memory_order_relaxed);
 }
 
 std::vector<std::string>

service/src/storage/document_store_lmdb.hpp

@@ -343,6 +343,15 @@ public:
     void set_relations(std::string_view collection, std::vector<RelationRef> rels);
     std::vector<RelationRef> relations(std::string_view collection);
 
+    // v2.11.0 final review (finding 5) — see DocumentStore::can_reject_writes.
+    //
+    // True iff `collection` has at least one declared unique field, or at
+    // least one declared relation with validateOnWrite (the only two things
+    // put() rejects a write for). The common case - nothing declared anywhere
+    // in the process - is answered by ONE relaxed atomic load with no mutex
+    // taken at all, which is what makes it usable on the hot write path.
+    bool can_reject_writes(std::string_view collection) override;
+
     // Outcome of the constructor's priming pass, for the owner to report.
     // prime_error() is empty on success.
     size_t primed_count() const noexcept { return primed_count_; }
@@ -485,6 +494,15 @@ private:
     // consults it on every put/del and must not contend with the dbi cache.
     std::mutex relations_mutex_;
     std::unordered_map<std::string, std::vector<RelationRef>> relations_;
+
+    // finding 5 — how many collections currently have a rejecting constraint
+    // declared (a unique field, or a relation with validateOnWrite), across
+    // BOTH maps. Maintained by set_unique_fields()/set_relations() so
+    // can_reject_writes() can answer "no, and nothing anywhere does" without
+    // taking either mutex. Never negative: each setter recomputes its own
+    // contribution rather than incrementing blindly.
+    std::atomic<size_t> rejecting_unique_collections_{0};
+    std::atomic<size_t> rejecting_relation_collections_{0};
     mutable std::atomic<uint64_t> indexed_scans_{0};
     mutable std::atomic<uint64_t> full_scans_{0};
     mutable std::atomic<uint64_t> declined_unselective_{0};

tests/load_test/test_client_namespacing.sh

@@ -5,9 +5,19 @@
 
 set -euo pipefail
 
-cd "$(dirname "$0")"
+# v2.11.0 final review (finding 10) — SCRIPT-RELATIVE, resolved ONCE and
+# BEFORE any cd.
+#
+# Three of these scripts hardcoded `ROOT=/data/smartbotic-database`, so running
+# them from a git worktree built and tested the MAIN checkout instead of the
+# branch under test: a green result was evidence about main, not about the
+# change. The two that were already script-relative computed ROOT from a
+# RELATIVE "$0" AFTER cd-ing, so `bash tests/load_test/<script>.sh` from the
+# repo root failed outright - they only worked when invoked by absolute path.
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+cd "$SCRIPT_DIR"
 
-ROOT=/data/smartbotic-database
 DIR=/tmp/sbdb-client-namespacing-e2e
 PORT=9011
 

tests/load_test/test_policy_enforcement.sh

@@ -8,9 +8,19 @@
 # real operator workflow.
 
 set -euo pipefail
-cd "$(dirname "$0")"
+# v2.11.0 final review (finding 10) — SCRIPT-RELATIVE, resolved ONCE and
+# BEFORE any cd.
+#
+# Three of these scripts hardcoded `ROOT=/data/smartbotic-database`, so running
+# them from a git worktree built and tested the MAIN checkout instead of the
+# branch under test: a green result was evidence about main, not about the
+# change. The two that were already script-relative computed ROOT from a
+# RELATIVE "$0" AFTER cd-ing, so `bash tests/load_test/<script>.sh` from the
+# repo root failed outright - they only worked when invoked by absolute path.
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+cd "$SCRIPT_DIR"
 
-ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
 DIR=/tmp/sbdb-policy-e2e
 PORT=9012
 

tests/load_test/test_relations.sh

@@ -92,6 +92,94 @@ echo
 echo "=== phase: verify declaration + write-path survived the restart ==="
 "$DRIVER" "127.0.0.1:$PORT" "$PROJECT_A" "$PROJECT_B" verify || fail "verify phase"
 
+# -------------------------------------------------------------------------
+# v2.11.0 final review (findings 1 and 9) — the surfaces the C++ client does
+# not expose. Client has no batchDelete()/dropCollection-with-relations and no
+# way to send a malformed on_delete, so these go over the raw wire.
+#
+# State at this point (established by the setup phase and carried across the
+# restart): relproj_a:workflows/wf-1 is a restrict-protected parent with
+# exactly one child, relproj_a:executions/ex-a1.
+# -------------------------------------------------------------------------
+GRPCURL="$(command -v grpcurl || true)"
+if [[ -z "$GRPCURL" ]]; then
+    fail "grpcurl is required for the BatchDelete/DropCollection/on_delete phase"
+fi
+rpc() {
+    local method="$1" body="$2"
+    "$GRPCURL" -plaintext -import-path "$REPO/proto" -proto database.proto \
+        -d "$body" "127.0.0.1:$PORT" "smartbotic.databasepb.DatabaseService/$method" 2>&1
+}
+
+echo
+echo "=== phase: BatchDelete must NOT bypass relation enforcement (finding 1) ==="
+OUT="$(rpc BatchDelete "{\"collection\":\"$PROJECT_A:workflows\",\"ids\":[\"wf-1\"]}" || true)"
+grep -q "FailedPrecondition" <<<"$OUT" \
+    || { echo "$OUT"; fail "BatchDelete of a restrict-protected parent was NOT refused"; }
+grep -q "wf_exec" <<<"$OUT" \
+    || { echo "$OUT"; fail "the BatchDelete refusal does not name the blocking relation"; }
+grep -q "nothing was deleted" <<<"$OUT" \
+    || { echo "$OUT"; fail "the BatchDelete refusal does not state that nothing was deleted"; }
+echo "  refused: $(head -2 <<<"$OUT" | tr '\n' ' ')"
+
+OUT="$(rpc Exists "{\"collection\":\"$PROJECT_A:workflows\",\"id\":\"wf-1\"}" || true)"
+grep -q '"exists": true' <<<"$OUT" \
+    || { echo "$OUT"; fail "the protected parent did not survive the refused BatchDelete"; }
+echo "  the parent is still there - the refusal mutated nothing"
+
+echo
+echo "=== phase: DropCollection must refuse a collection in a relation (finding 1 audit) ==="
+OUT="$(rpc DropCollection "{\"name\":\"$PROJECT_A:workflows\"}" || true)"
+grep -q "FailedPrecondition" <<<"$OUT" \
+    || { echo "$OUT"; fail "DropCollection on a relation parent was NOT refused"; }
+grep -q "participates in" <<<"$OUT" \
+    || { echo "$OUT"; fail "the DropCollection refusal does not explain why"; }
+OUT="$(rpc Exists "{\"collection\":\"$PROJECT_A:workflows\",\"id\":\"wf-1\"}" || true)"
+grep -q '"exists": true' <<<"$OUT" \
+    || { echo "$OUT"; fail "DropCollection destroyed the collection despite refusing"; }
+echo "  refused, and the collection is intact"
+
+echo
+echo "=== phase: an unrecognised on_delete is REFUSED, not coerced (finding 9) ==="
+OUT="$(rpc CreateRelation "{\"name\":\"$PROJECT_A:typo_rel\",\"child\":\"$PROJECT_A:executions\",\"child_field\":\"workflowId\",\"parent\":\"$PROJECT_A:workflows\",\"on_delete\":\"Cascade\"}" || true)"
+grep -q "on_delete must be one of" <<<"$OUT" \
+    || { echo "$OUT"; fail "a typo'd on_delete was silently coerced instead of refused"; }
+grep -q "Cascade" <<<"$OUT" \
+    || { echo "$OUT"; fail "the refusal does not echo the value that was rejected"; }
+OUT="$(rpc GetRelationInfo "{\"name\":\"$PROJECT_A:typo_rel\"}" || true)"
+grep -q '"found": true' <<<"$OUT" \
+    && { echo "$OUT"; fail "the refused relation was persisted anyway"; }
+echo "  refused, and nothing was declared"
+
+echo
+echo "=== phase: BatchDelete still deletes what it is allowed to (finding 1) ==="
+OUT="$(rpc BatchDelete "{\"collection\":\"$PROJECT_A:executions\",\"ids\":[\"ex-a1\"]}" || true)"
+grep -q '"deletedCount": "1"' <<<"$OUT" \
+    || { echo "$OUT"; fail "BatchDelete of an unprotected child did not delete it"; }
+OUT="$(rpc BatchDelete "{\"collection\":\"$PROJECT_A:workflows\",\"ids\":[\"wf-1\"]}" || true)"
+grep -q '"deletedCount": "1"' <<<"$OUT" \
+    || { echo "$OUT"; fail "BatchDelete still refused the parent after its last child was removed"; }
+echo "  the child went first, then the parent - enforcement, not a blanket refusal"
+
+# Put the fixture back for the self-heal phase below, which expects one child.
+# ⚠ UpsertRequest.data is `bytes`, so grpcurl needs it BASE64-ENCODED. Passing
+# the raw JSON string there is accepted and silently stores nothing, which is
+# exactly what the first attempt did.
+b64() { printf '%s' "$1" | base64 -w0; }
+PARENT_B64="$(b64 '{"name":"wf-1"}')"
+CHILD_B64="$(b64 '{"workflowId":"wf-1"}')"
+rpc Upsert "{\"collection\":\"$PROJECT_A:workflows\",\"id\":\"wf-1\",\"data\":\"$PARENT_B64\"}" >/dev/null
+rpc Upsert "{\"collection\":\"$PROJECT_A:executions\",\"id\":\"ex-a1\",\"data\":\"$CHILD_B64\"}" >/dev/null
+OUT="$(rpc Exists "{\"collection\":\"$PROJECT_A:executions\",\"id\":\"ex-a1\"}" || true)"
+grep -q '"exists": true' <<<"$OUT" \
+    || { echo "$OUT"; fail "could not restore the e2e fixture for the self-heal phase"; }
+# And the reverse-index posting must be back, or the self-heal phase below is
+# asserting against an empty index and would pass for the wrong reason.
+OUT="$(rpc Delete "{\"collection\":\"$PROJECT_A:workflows\",\"id\":\"wf-1\"}" || true)"
+grep -q "FailedPrecondition" <<<"$OUT" \
+    || { echo "$OUT"; fail "the restored child is not indexed - the fixture is not equivalent"; }
+echo "  fixture restored, and the restored child is indexed"
+
 echo
 echo "=== manufacturing a missing reverse-index sub-db (service stopped) ==="
 stop_server

tests/load_test/test_relations_client_e2e.sh

@@ -5,9 +5,19 @@
 
 set -euo pipefail
 
-cd "$(dirname "$0")"
+# v2.11.0 final review (finding 10) — SCRIPT-RELATIVE, resolved ONCE and
+# BEFORE any cd.
+#
+# Three of these scripts hardcoded `ROOT=/data/smartbotic-database`, so running
+# them from a git worktree built and tested the MAIN checkout instead of the
+# branch under test: a green result was evidence about main, not about the
+# change. The two that were already script-relative computed ROOT from a
+# RELATIVE "$0" AFTER cd-ing, so `bash tests/load_test/<script>.sh` from the
+# repo root failed outright - they only worked when invoked by absolute path.
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+cd "$SCRIPT_DIR"
 
-ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
 DIR=/tmp/sbdb-relations-client-e2e
 PORT=9012
 

tests/load_test/test_v24_tls_auth.sh

@@ -14,9 +14,19 @@
 
 set -euo pipefail
 
-cd "$(dirname "$0")"
+# v2.11.0 final review (finding 10) — SCRIPT-RELATIVE, resolved ONCE and
+# BEFORE any cd.
+#
+# Three of these scripts hardcoded `ROOT=/data/smartbotic-database`, so running
+# them from a git worktree built and tested the MAIN checkout instead of the
+# branch under test: a green result was evidence about main, not about the
+# change. The two that were already script-relative computed ROOT from a
+# RELATIVE "$0" AFTER cd-ing, so `bash tests/load_test/<script>.sh` from the
+# repo root failed outright - they only worked when invoked by absolute path.
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+cd "$SCRIPT_DIR"
 
-ROOT=/data/smartbotic-database
 DIR=/tmp/sbdbv24-tls-auth-e2e
 
 rm -rf "$DIR"

tests/load_test/test_views_multiproject.sh

@@ -20,9 +20,19 @@
 
 set -euo pipefail
 
-cd "$(dirname "$0")"
+# v2.11.0 final review (finding 10) — SCRIPT-RELATIVE, resolved ONCE and
+# BEFORE any cd.
+#
+# Three of these scripts hardcoded `ROOT=/data/smartbotic-database`, so running
+# them from a git worktree built and tested the MAIN checkout instead of the
+# branch under test: a green result was evidence about main, not about the
+# change. The two that were already script-relative computed ROOT from a
+# RELATIVE "$0" AFTER cd-ing, so `bash tests/load_test/<script>.sh` from the
+# repo root failed outright - they only worked when invoked by absolute path.
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+cd "$SCRIPT_DIR"
 
-ROOT=/data/smartbotic-database
 DIR=/tmp/sbdb-views-multiproject
 PORT=9078
 

tests/test_relation_enforcement.cpp

@@ -32,6 +32,8 @@
 #include "storage/document_store_lmdb.hpp"
 #include "storage/lmdb_env.hpp"
 
+#include <lmdb.h>
+
 namespace fs = std::filesystem;
 
 using smartbotic::database::CascadeBlocked;
@@ -1047,10 +1049,15 @@ void test_replayed_cascade_update_remirrors_to_lmdb_after_crash_window() {
     cfg2.dataDir = p.path;
     PersistenceManager pm2(cfg2);
     auto outcome = pm2.recover(freshStore);
+    // v2.11.0 final review (finding 4) — recover() only COLLECTS the list now;
+    // DatabaseService::initialize() runs the pass after
+    // applyRelationDeclarations()/applyIndexDeclarations(). Mirrored here, in
+    // that order, so the test drives the production sequence.
+    pm2.runPendingRemirror(freshStore, outcome);
     check(outcome.kind != smartbotic::database::RecoveryOutcome::Kind::Failed,
           "recovery did not fail");
     check(outcome.updatesRemirroredAfterReplay >= 1,
-          "recover() re-mirrored at least the node's replayed UPDATE");
+          "the re-mirror pass handled at least the node's replayed UPDATE");
 
     // MemoryStore converges (this part already worked before this fix).
     auto memNode = freshStore.get("default:nodes", "n1");
@@ -1139,6 +1146,7 @@ void test_a_failing_row_does_not_stop_recovery() {
     cfg2.dataDir = p.path;
     PersistenceManager pm2(cfg2);
     auto outcome = pm2.recover(freshStore);
+    pm2.runPendingRemirror(freshStore, outcome);   // finding 4 — see above
 
     // THE finding: recovery completes.
     check(outcome.kind != smartbotic::database::RecoveryOutcome::Kind::Failed,
@@ -1205,6 +1213,7 @@ void test_reinsert_after_delete_converges_in_lmdb() {
     cfg2.dataDir = p.path;
     PersistenceManager pm2(cfg2);
     auto outcome = pm2.recover(freshStore);
+    pm2.runPendingRemirror(freshStore, outcome);   // finding 4 — see above
     check(outcome.kind != smartbotic::database::RecoveryOutcome::Kind::Failed, "recovery completed");
     check(outcome.updatesRemirroredAfterReplay == 1,
           "the reinserted id was re-mirrored (INSERT is tracked, not just UPDATE/UPSERT)");
@@ -1650,6 +1659,426 @@ void test_validate_on_write_closes_the_stale_check_race() {
 
 }  // namespace
 
+
+// =========================================================================
+// v2.11.0 final review — the post-replay re-mirror pass runs AFTER arming
+// (finding 4) and its window is bounded by bytes as well as count (finding 3).
+// =========================================================================
+
+// FINDING 4, the consequential half: the pass writes documents through
+// LmdbDocumentStore::put(), which maintains every declared secondary index
+// INSIDE the document's own write transaction by reading
+// indexed_fields(collection) live. While the pass ran inside recover() - line
+// ~88 of DatabaseService::initialize(), where applyIndexDeclarations() is line
+// ~208 - that map was still EMPTY, so the pass moved the document forward and
+// left its postings behind. An index-served query then returns the row under
+// its OLD value and misses it under its new one: silently wrong rows, which is
+// exactly what putting maintainIndexes() inside the transaction exists to
+// prevent, and which bites ANY install with a v2.9 index declared, relations
+// or not.
+//
+// The simulated restart clears the in-process declaration (a real restart
+// rebuilds it from `_collection_meta`) and re-arms it only AFTER recover(),
+// which is the production ordering. Moving the runPendingRemirror() call above
+// the re-arm reproduces the bug.
+void test_remirror_maintains_the_secondary_index() {
+    TmpEnv t("rel-remirror-index");
+    LmdbDocumentStore store(t.env);
+    TmpPersistence p("rel-remirror-index-wal");
+    check(p.pm.start(), "persistence manager started");
+
+    store.set_indexed_fields("widgets", {"status"});
+
+    Document d; d.id = "w1"; d.collection = "widgets";
+    d.set_data({{"status", "queued"}});
+    store.put("widgets", "w1", d);            // LMDB + posting under "queued"
+    p.pm.logInsert("default:widgets", d);
+
+    {
+        auto pre = store.index_lookup_eq("widgets", "status", nlohmann::json("queued"));
+        check(pre.has_value() && pre->size() == 1, "the pre-crash posting exists under 'queued'");
+    }
+
+    // The crash window: a WAL UPDATE whose LMDB write never ran.
+    d.set_data({{"status", "done"}});
+    p.pm.logUpdate("default:widgets", d);
+    p.pm.stop();
+
+    // "Restart". The declaration is in-process state, so clear it: a real
+    // restart starts with an empty map and rebuilds it in
+    // applyIndexDeclarations().
+    store.set_indexed_fields("widgets", {});
+
+    MemoryStore freshStore(MemoryStore::Config{});
+    freshStore.start();
+    std::atomic<bool> mirrorHealthy{true};
+    std::atomic<uint64_t> mirrorDrift{0};
+    freshStore.setDocumentStoreMirror(
+        [&store](std::string_view) -> smartbotic::db::storage::DocumentStore* { return &store; },
+        &mirrorHealthy, &mirrorDrift);
+
+    PersistenceManager::Config cfg2;
+    cfg2.dataDir = p.path;
+    PersistenceManager pm2(cfg2);
+    auto outcome = pm2.recover(freshStore);
+    check(outcome.kind != smartbotic::database::RecoveryOutcome::Kind::Failed, "recovery completed");
+    check(outcome.pendingRemirror.size() == 1,
+          "recover() COLLECTED the row instead of re-mirroring it itself");
+
+    // applyIndexDeclarations()' place in initialize(): before the pass.
+    store.set_indexed_fields("widgets", {"status"});
+    pm2.runPendingRemirror(freshStore, outcome);
+    check(outcome.updatesRemirroredAfterReplay == 1, "the row was re-mirrored");
+
+    // The document converged (this half worked before the fix too).
+    auto lmdb = store.get("widgets", "w1");
+    check(lmdb.has_value() && lmdb->data()["status"] == "done",
+          "LMDB holds the replayed value");
+
+    // LOAD-BEARING: the INDEX converged with it. Both directions matter - a
+    // stale posting under the old value is a wrong row returned, and a missing
+    // posting under the new value is a row silently omitted.
+    auto stale = store.index_lookup_eq("widgets", "status", nlohmann::json("queued"));
+    check(stale.has_value() && stale->empty(),
+          "no posting left under the OLD value - an index-served query cannot "
+          "return this row as if it were still queued");
+    auto fresh = store.index_lookup_eq("widgets", "status", nlohmann::json("done"));
+    check(fresh.has_value() && fresh->size() == 1 && (*fresh)[0] == "w1",
+          "the row IS found under its new value - the posting was maintained by "
+          "the pass, which is only possible because the index was armed first");
+
+    freshStore.stop();
+}
+
+// FINDING 4, the other half: with the pass running after arming, a
+// UniqueViolation from it is REACHABLE, which is what makes phase 3's single
+// retry earn its place. It could not fire at all while the pass ran inside
+// recover() (unique_fields_ was empty), so the retry, its "moved unique value"
+// rationale and persistence_manager.cpp's insertion-ORDERED comment were all
+// dead code justified by dead reasoning.
+//
+// The scenario, in WAL order: a NEW row b1 takes the email a1 currently holds,
+// and only then does a1 move off it. Re-mirroring in WAL order therefore tries
+// b1 first, against a1's not-yet-updated posting - a genuine conflict - and it
+// clears only once a1 has been re-mirrored. Replacing the retry with a bare
+// noteFailure() makes this test fail.
+void test_moved_unique_value_is_resolved_by_the_retry_pass() {
+    TmpEnv t("rel-remirror-unique");
+    LmdbDocumentStore store(t.env);
+    TmpPersistence p("rel-remirror-unique-wal");
+    check(p.pm.start(), "persistence manager started");
+
+    store.set_indexed_fields("people", {"email"});
+    store.set_unique_fields("people", {"email"});
+
+    // Pre-crash LMDB state: a1 holds the value, with its posting. Written
+    // straight to LMDB and deliberately NOT to the WAL - it is the state a
+    // snapshot would have carried, and keeping it out of the WAL is what puts
+    // b1 first in the re-mirror order.
+    Document a; a.id = "a1"; a.collection = "people";
+    a.set_data({{"email", "shared@example.com"}});
+    store.put("people", "a1", a);
+
+    // WAL order: b1 claims the value FIRST, a1 vacates it second. That order is
+    // what creates the transient conflict, and it is the order recover()
+    // preserves (insertion-ORDERED, deliberately - see persistence_manager.cpp).
+    Document b; b.id = "b1"; b.collection = "people";
+    b.set_data({{"email", "shared@example.com"}});
+    p.pm.logInsert("default:people", b);
+
+    a.set_data({{"email", "moved@example.com"}});
+    p.pm.logInsert("default:people", a);
+    p.pm.stop();
+
+    store.set_unique_fields("people", {});
+    store.set_indexed_fields("people", {});
+
+    MemoryStore freshStore(MemoryStore::Config{});
+    freshStore.start();
+    std::atomic<bool> mirrorHealthy{true};
+    std::atomic<uint64_t> mirrorDrift{0};
+    freshStore.setDocumentStoreMirror(
+        [&store](std::string_view) -> smartbotic::db::storage::DocumentStore* { return &store; },
+        &mirrorHealthy, &mirrorDrift);
+
+    PersistenceManager::Config cfg2;
+    cfg2.dataDir = p.path;
+    PersistenceManager pm2(cfg2);
+    auto outcome = pm2.recover(freshStore);
+    check(outcome.pendingRemirror.size() == 2, "both ids collected, in WAL order");
+    check(outcome.pendingRemirror[0].second == "b1",
+          "b1 comes FIRST - the order is what creates the transient conflict");
+
+    store.set_indexed_fields("people", {"email"});
+    store.set_unique_fields("people", {"email"});
+    // chunkSize=1 so each row gets its own transaction: with a batch of two the
+    // whole chunk would abort and be retried row by row anyway, but forcing the
+    // per-row shape makes the deferral the pass's own, not put_batch's.
+    auto res = freshStore.remirrorDocuments(outcome.pendingRemirror, /*chunkSize=*/1);
+
+    check(res.remirrored == 2,
+          "BOTH rows converged - b1 was deferred on the conflict and committed by "
+          "the retry once a1 had vacated the value");
+    check(res.failed == 0, "nothing was written off as failed");
+    check(mirrorDrift.load() == 0, "no drift bumped - a resolved conflict is not drift");
+
+    auto la = store.get("people", "a1");
+    auto lb = store.get("people", "b1");
+    check(la.has_value() && la->data()["email"] == "moved@example.com", "a1 moved off the value");
+    check(lb.has_value() && lb->data()["email"] == "shared@example.com", "b1 holds the value now");
+    auto who = store.index_lookup_eq("people", "email", nlohmann::json("shared@example.com"));
+    check(who.has_value() && who->size() == 1 && (*who)[0] == "b1",
+          "exactly one row holds the unique value, and it is the right one");
+
+    freshStore.stop();
+}
+
+// FINDING 3: the window is bounded by BYTES as well as by count, and a window
+// boundary is crossed at all (no prior test did - every fixture fitted inside
+// one chunk of 256, so the streaming structure round 5 introduced was
+// unpinned).
+//
+// chunkBytes=1 forces one row per window, which is the extreme the byte budget
+// can reach; every row must still converge, and the loop must terminate. A
+// document larger than the whole budget forms a window of one rather than
+// stalling, which is what guarantees progress.
+void test_remirror_windows_are_bounded_by_bytes_and_by_count() {
+    TmpEnv t("rel-remirror-window");
+    LmdbDocumentStore store(t.env);
+    TmpPersistence p("rel-remirror-window-wal");
+    check(p.pm.start(), "persistence manager started");
+
+    constexpr int kRows = 7;
+    for (int i = 0; i < kRows; ++i) {
+        Document d; d.id = "w" + std::to_string(i); d.collection = "widgets";
+        d.set_data({{"n", i}, {"pad", std::string(4096, 'x')}});
+        p.pm.logInsert("default:widgets", d);
+    }
+    p.pm.stop();
+
+    MemoryStore freshStore(MemoryStore::Config{});
+    freshStore.start();
+    std::atomic<bool> mirrorHealthy{true};
+    std::atomic<uint64_t> mirrorDrift{0};
+    freshStore.setDocumentStoreMirror(
+        [&store](std::string_view) -> smartbotic::db::storage::DocumentStore* { return &store; },
+        &mirrorHealthy, &mirrorDrift);
+
+    PersistenceManager::Config cfg2;
+    cfg2.dataDir = p.path;
+    PersistenceManager pm2(cfg2);
+    auto outcome = pm2.recover(freshStore);
+    check(outcome.pendingRemirror.size() == kRows, "all rows collected");
+
+    // Byte budget of 1 with a generous count cap: the BYTES must be what closes
+    // each window. Before this fix chunkBytes did not exist and the count cap
+    // alone put all seven in one window - on this repo's ~2.9 MB documents, 256
+    // of them is ~750 MB resident plus as much again in LMDB dirty pages, in
+    // the happy path, on the boot path.
+    //
+    // The OBSERVABLE is the number of committed LMDB transactions, read from
+    // the env's own last-txnid. Row counts alone cannot discriminate here (all
+    // seven rows converge either way, via put_batch instead of put), so
+    // counting commits is what actually pins that the window closed per row.
+    auto lastTxnId = [&]() -> uint64_t {
+        MDB_envinfo info;
+        mdb_env_info(t.env.raw(), &info);
+        return static_cast<uint64_t>(info.me_last_txnid);
+    };
+    const uint64_t txnBefore = lastTxnId();
+    auto res = freshStore.remirrorDocuments(outcome.pendingRemirror,
+                                           /*chunkSize=*/1024, /*chunkBytes=*/1);
+    const uint64_t commits = lastTxnId() - txnBefore;
+    check(res.remirrored == kRows,
+          "every row converged with a byte budget smaller than one document - "
+          "an oversized row forms a window of one rather than stalling");
+    check(res.failed == 0, "no failures");
+    check(commits >= kRows,
+          "the BYTE budget closed each window: one commit per row, not one "
+          "commit for all seven (which is what the count cap alone produced, "
+          "and what makes 256 rows of 2.9 MB a ~750 MB resident window)");
+    for (int i = 0; i < kRows; ++i) {
+        auto got = store.get("widgets", "w" + std::to_string(i));
+        check(got.has_value() && got->data()["n"] == i, "row survived its own window");
+    }
+
+    // And the count cap still closes a window on its own: two rows, chunkSize=1.
+    LmdbDocumentStore store2(t.env);
+    freshStore.setDocumentStoreMirror(
+        [&store2](std::string_view) -> smartbotic::db::storage::DocumentStore* { return &store2; },
+        &mirrorHealthy, &mirrorDrift);
+    auto res2 = freshStore.remirrorDocuments(outcome.pendingRemirror,
+                                             /*chunkSize=*/1,
+                                             /*chunkBytes=*/64ull * 1024 * 1024);
+    check(res2.remirrored == kRows, "count-bounded windows cover every row too");
+    check(res2.failed == 0, "no failures crossing count-bounded window boundaries");
+
+    freshStore.stop();
+}
+
+// FINDING 6: the cascade path builds every updatedDoc from the LMDB copy and
+// pushes it into MemoryStore. While the mirror is unhealthy or drifted LMDB may
+// be BEHIND - that is the entire reason reads fall back to MemoryStore - so the
+// cascade would overwrite MemoryStore's fresher child body with a stale one.
+// Real data loss, on a state this codebase treats as routine.
+void test_cascade_refuses_while_the_mirror_is_unhealthy_or_drifted() {
+    TmpEnv t("rel-cascade-gate");
+    LmdbDocumentStore store(t.env);
+    TmpPersistence p("rel-cascade-gate-wal");
+    check(p.pm.start(), "persistence manager started");
+
+    MemoryStore mstore(MemoryStore::Config{});
+    mstore.start();
+    RelationManager rm(mstore);
+    rm.loadFromStore();
+    CollectionConfigManager cfgManager(mstore);
+
+    std::atomic<bool> healthy{true};
+    std::atomic<uint64_t> drift{0};
+    mstore.setDocumentStoreMirror(
+        [&store](std::string_view) -> smartbotic::db::storage::DocumentStore* { return &store; },
+        &healthy, &drift);
+
+    RelationInfo rel;
+    rel.name = "default:exec_wf";
+    rel.child = "default:executions";
+    rel.childField = "workflowId";
+    rel.parent = "default:workflows";
+    rel.onDelete = OnDelete::Cascade;
+    std::string mgrErr;
+    check(rm.createRelation(rel, mgrErr), "declared the cascade relation");
+    store.set_relations("executions",
+                        {RelationRef{"exec_wf", "workflowId", "workflows", false, true}});
+
+    Document parent; parent.id = "wf-1"; parent.collection = "workflows";
+    parent.set_data({{"name", "wf-1"}});
+    store.put("workflows", "wf-1", parent);
+    mstore.loadDocument("default:workflows", parent);
+
+    Document child; child.id = "ex-1"; child.collection = "executions";
+    child.set_data({{"workflowId", "wf-1"}});
+    store.put("executions", "ex-1", child);
+    mstore.loadDocument("default:executions", child);
+    check(store.relation_index_child_count("exec_wf", "wf-1") == 1, "one child indexed");
+
+    auto expectRefusal = [&](const char* what) {
+        bool refused = false;
+        std::string msg;
+        try {
+            executeCascade(rm, store, p.pm, mstore, cfgManager, "default:workflows", "wf-1", nullptr);
+        } catch (const CascadeBlocked& e) {
+            refused = true;
+            msg = e.what();
+        } catch (const std::exception& e) {
+            msg = std::string("wrong exception type: ") + e.what();
+        }
+        check(refused, what);
+        check(msg.find("LMDB mirror is") != std::string::npos,
+              "the refusal tells the operator WHY, not just that it failed");
+        // Nothing may have been touched anywhere: the gate runs before
+        // planCascade(), so before writeCascadeWal() and before any LMDB commit.
+        check(store.get("workflows", "wf-1").has_value(), "parent untouched in LMDB");
+        check(store.get("executions", "ex-1").has_value(), "child untouched in LMDB");
+        check(mstore.get("default:workflows", "wf-1").has_value(), "parent untouched in MemoryStore");
+        check(mstore.get("default:executions", "ex-1").has_value(), "child untouched in MemoryStore");
+    };
+
+    healthy.store(false);
+    expectRefusal("cascade refused while the mirror is UNHEALTHY");
+
+    healthy.store(true);
+    drift.store(1);
+    expectRefusal("cascade refused while the mirror has DRIFTED (health alone is not enough - "
+                  "the re-mirror pass bumps drift without flipping health, by design)");
+
+    // And with a clean mirror it proceeds, so the gate is not a blanket refusal.
+    drift.store(0);
+    bool parentExisted = false;
+    try {
+        parentExisted = executeCascade(rm, store, p.pm, mstore, cfgManager,
+                                       "default:workflows", "wf-1", nullptr);
+    } catch (const std::exception& e) {
+        check(false, "cascade threw with a healthy mirror");
+        std::cerr << "  (" << e.what() << ")\n";
+    }
+    check(parentExisted, "the cascade ran once the mirror was healthy and undrifted");
+    check(!store.get("executions", "ex-1").has_value(), "child cascaded away");
+    check(!store.get("workflows", "wf-1").has_value(), "parent deleted");
+
+    mstore.stop();
+}
+
+
+// FINDING 5: the gate that keeps T11's per-write Document deep copy off the
+// write path of every install that declares no rejecting constraint.
+//
+// Measured (this repo's shapes, same machine): the copy alone is 1.8 us at
+// 4 KB, 7.1 us at 64 KB and 364 us at 2.9 MB, all of it inside the
+// collection's unique_lock and all of it a fresh yyjson_mut_doc allocation.
+// The gate's own cost in the common case is one relaxed atomic load.
+//
+// The truth table is what matters, and one row is easy to get wrong: a
+// relation WITHOUT validateOnWrite maintains its reverse index on every put
+// but can never THROW, so it must not force a snapshot.
+void test_can_reject_writes_gates_the_undo_snapshot() {
+    TmpEnv t("rel-gate-undo");
+    LmdbDocumentStore store(t.env);
+
+    check(!store.can_reject_writes("widgets"),
+          "nothing declared anywhere: no snapshot needed (the case that used to "
+          "pay for the copy on every install)");
+
+    store.set_indexed_fields("widgets", {"status"});
+    check(!store.can_reject_writes("widgets"),
+          "a plain secondary index cannot reject a write, so still no snapshot");
+
+    store.set_relations("widgets",
+                        {RelationRef{"w_rel", "parentId", "parents", /*validateOnWrite=*/false,
+                                     /*relationsEnforced=*/true}});
+    check(!store.can_reject_writes("widgets"),
+          "a relation WITHOUT validate_on_write maintains the reverse index but "
+          "never throws - no snapshot");
+
+    store.set_relations("widgets",
+                        {RelationRef{"w_rel", "parentId", "parents", /*validateOnWrite=*/true,
+                                     /*relationsEnforced=*/true}});
+    check(store.can_reject_writes("widgets"),
+          "validate_on_write CAN reject (MissingParentReference) - snapshot needed");
+    check(!store.can_reject_writes("others"),
+          "and it is per collection, not process-wide: an unrelated collection "
+          "still pays nothing");
+
+    store.set_relations("widgets", {});
+    check(!store.can_reject_writes("widgets"), "dropping the relation drops the need");
+
+    store.set_unique_fields("widgets", {"email"});
+    check(store.can_reject_writes("widgets"),
+          "a unique field CAN reject (UniqueViolation) - snapshot needed");
+    check(!store.can_reject_writes("others"), "still per collection");
+    store.set_unique_fields("widgets", {});
+    check(!store.can_reject_writes("widgets"), "and dropping it drops the need again");
+
+    // MemoryStore's view of the same question, which is what the write paths
+    // actually call. Fails SAFE when no mirror is wired.
+    MemoryStore ms(MemoryStore::Config{});
+    ms.start();
+    check(ms.needsUndoSnapshot("default:widgets"),
+          "with NO mirror wired the answer is 'take the snapshot' - fail safe");
+    std::atomic<bool> healthy{true};
+    std::atomic<uint64_t> drift{0};
+    ms.setDocumentStoreMirror(
+        [&store](std::string_view) -> smartbotic::db::storage::DocumentStore* { return &store; },
+        &healthy, &drift);
+    check(!ms.needsUndoSnapshot("default:widgets"),
+          "wired, with nothing declared: no snapshot");
+    store.set_unique_fields("widgets", {"email"});
+    check(ms.needsUndoSnapshot("default:widgets"),
+          "wired, with a unique field declared: snapshot");
+    check(ms.needsUndoSnapshot("this is not a valid:collection:name"),
+          "an unparseable name fails safe too");
+    ms.stop();
+}
+
 int main() {
     std::cout << "=== test_relation_enforcement ===\n";
     test_index_follows_the_child_field();
@@ -1678,6 +2107,11 @@ int main() {
     test_validate_on_write_rolls_back_an_already_applied_sibling_relation();
     test_validate_on_write_relations_enforced_false_is_the_escape_hatch();
     test_validate_on_write_closes_the_stale_check_race();
+    test_remirror_maintains_the_secondary_index();
+    test_moved_unique_value_is_resolved_by_the_retry_pass();
+    test_remirror_windows_are_bounded_by_bytes_and_by_count();
+    test_cascade_refuses_while_the_mirror_is_unhealthy_or_drifted();
+    test_can_reject_writes_gates_the_undo_snapshot();
 
     std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
     return g_fail == 0 ? 0 : 1;

tests/test_relation_manager.cpp

@@ -154,12 +154,191 @@ void test_load_skips_a_cross_project_relation_written_by_hand() {
 
 }  // namespace
 
+
+// =========================================================================
+// v2.11.0 final review, finding 8 — the bare-vs-qualified bug class, made
+// unrepresentable at the RelationManager boundary rather than spot-fixed at
+// the caller.
+//
+// The live bug: armRelationsForChild() looked relations up under the caller's
+// RAW string while boot arming used the canonical "<project>:<collection>",
+// and set_relations() keys the storage map by the BARE name either way. So a
+// raw-gRPC caller naming "executions" instead of "default:executions" found
+// zero relations, and set_relations(bare, {}) then ERASED the entry the
+// canonical arming had filled - disarming both the reverse index and
+// validate_on_write for the life of the process, logged only as "re-armed 0
+// relation(s)", and silently repaired by a restart.
+//
+// This is the third occurrence of the class (v2.4.2 lost every view for two
+// releases, v2.4.5 gave every collection a phantom twin), so the fix is at the
+// one funnel every entry point already goes through.
+void test_bare_and_qualified_names_are_the_same_relation() {
+    Fixture f;
+    RelationManager rm(f.store);
+    rm.loadFromStore();
+
+    // Declared with an UNQUALIFIED name and unqualified endpoints, exactly as a
+    // raw-gRPC caller (or a hand-written migration) would.
+    RelationInfo bare;
+    bare.name = "exec_wf";
+    bare.child = "executions";
+    bare.childField = "workflowId";
+    bare.parent = "workflows";
+    std::string err;
+    check(rm.createRelation(bare, err), "a bare-named relation is accepted");
+
+    // It is STORED canonically, so everything downstream sees one form.
+    auto viaBare = rm.getRelation("exec_wf");
+    check(viaBare.has_value(), "found under the bare name the caller used");
+    check(viaBare && viaBare->name == "default:exec_wf",
+          "but it reports its CANONICAL name - the declaration was normalised, "
+          "not stored verbatim");
+    check(viaBare && viaBare->child == "default:executions", "child canonicalised too");
+    check(viaBare && viaBare->parent == "default:workflows", "parent canonicalised too");
+
+    auto viaQualified = rm.getRelation("default:exec_wf");
+    check(viaQualified.has_value(), "and found under the qualified name as well");
+
+    // A second declaration under the OTHER spelling is a DUPLICATE, not a new
+    // relation. Before the fix this created a second entry that armed the same
+    // bare collection and clobbered the first.
+    RelationInfo qualified = bare;
+    qualified.name = "default:exec_wf";
+    qualified.child = "default:executions";
+    qualified.parent = "default:workflows";
+    check(!rm.createRelation(qualified, err),
+          "declaring the qualified spelling of an existing bare relation is refused "
+          "as a duplicate");
+    check(err.find("already exists") != std::string::npos, "and says why");
+    check(rm.listRelations().size() == 1, "still exactly one relation, not two");
+
+    // THE LOOKUP THAT WAS BROKEN: armRelationsForChild's, and Delete's.
+    check(rm.relationsWithChild("executions").size() == 1,
+          "relationsWithChild finds it under the BARE collection name - this is "
+          "the lookup that returned zero and caused the disarm");
+    check(rm.relationsWithChild("default:executions").size() == 1,
+          "and under the qualified one");
+    check(rm.relationsWithParent("workflows").size() == 1,
+          "relationsWithParent likewise - a bare name here would report 'nobody's "
+          "parent' and permit every delete");
+    check(rm.relationsWithParent("default:workflows").size() == 1, "and qualified");
+
+    // Cross-project isolation is not weakened by canonicalisation.
+    check(rm.relationsWithChild("acme:executions").empty(),
+          "another project's same-named collection still matches nothing");
+
+    // Dropping by either spelling works, and actually removes the stored row.
+    check(rm.dropRelation("exec_wf", err), "droppable by the bare name");
+    check(!rm.getRelation("default:exec_wf").has_value(), "gone from the cache");
+    RelationManager reloaded(f.store);
+    reloaded.loadFromStore();
+    check(reloaded.listRelations().empty(),
+          "and gone from the store - dropping by the bare name really removed the "
+          "canonical document, it did not just clear the cache");
+}
+
+// finding 8, the migration half: a record already stored under a non-canonical
+// document id is re-keyed on load, in the store as well as in the cache.
+// Without the store half, dropRelation() (which removes by canonical name)
+// would leave the row behind and the relation would come back on the next boot.
+void test_load_rekeys_a_legacy_bare_named_record() {
+    Fixture f;
+    // Hand-write a record the way a pre-fix createRelation would have stored it.
+    Document d;
+    d.id = "legacy_rel";
+    d.set_data(nlohmann::json{
+        {"name", "legacy_rel"},
+        {"child", "executions"},
+        {"child_field", "workflowId"},
+        {"parent", "workflows"},
+        {"on_delete", "restrict"},
+        {"validate_on_write", false}});
+    f.store.createCollection(RelationManager::SYSTEM_COLLECTION, CollectionOptions{});
+    f.store.upsert(RelationManager::SYSTEM_COLLECTION, d);
+
+    RelationManager rm(f.store);
+    rm.loadFromStore();
+    auto got = rm.getRelation("default:legacy_rel");
+    check(got.has_value(), "the legacy record loaded");
+    check(got && got->name == "default:legacy_rel", "under its canonical name");
+
+    // The STORE was re-keyed, not just the cache.
+    check(!f.store.get(RelationManager::SYSTEM_COLLECTION, "legacy_rel").has_value(),
+          "the old bare-keyed document is gone");
+    check(f.store.get(RelationManager::SYSTEM_COLLECTION, "default:legacy_rel").has_value(),
+          "and a canonically-keyed one exists - so dropRelation(), which removes "
+          "by the canonical name, can actually remove it");
+
+    std::string err;
+    check(rm.dropRelation("default:legacy_rel", err), "and it drops");
+    RelationManager reloaded(f.store);
+    reloaded.loadFromStore();
+    check(reloaded.listRelations().empty(), "staying dropped across a reload");
+}
+
+// finding 9 — an unrecognised on_delete must be REFUSED, not coerced to
+// Restrict. There used to be two copies of the parser (one in
+// database_grpc_impl.cpp, one here) and both coerced silently. That failed safe
+// while cascade/set_null were inert; T12 made them destructive in the other
+// direction, so an operator who typed "Cascade" was told the relation was
+// created and believed cascade was armed while restrict was.
+void test_on_delete_is_validated_not_coerced() {
+    check(parseOnDelete("restrict").has_value(), "restrict parses");
+    check(parseOnDelete("cascade") == OnDelete::Cascade, "cascade parses");
+    check(parseOnDelete("set_null") == OnDelete::SetNull, "set_null parses");
+    check(parseOnDelete("no_action") == OnDelete::NoAction, "no_action parses");
+
+    // The typo cases that used to become Restrict silently.
+    check(!parseOnDelete("Cascade").has_value(), "'Cascade' is REFUSED, not coerced");
+    check(!parseOnDelete("CASCADE").has_value(), "'CASCADE' is refused");
+    check(!parseOnDelete("cascde").has_value(), "a misspelling is refused");
+    check(!parseOnDelete("setnull").has_value(), "'setnull' is refused");
+    check(!parseOnDelete("").has_value(),
+          "and the empty string is refused HERE - the RPC layer, not the parser, "
+          "is what maps absent to the documented default");
+
+    // Round-trips through the one renderer, so the two directions cannot drift.
+    for (auto v : {OnDelete::Restrict, OnDelete::Cascade, OnDelete::SetNull,
+                   OnDelete::NoAction}) {
+        check(parseOnDelete(onDeleteToString(v)) == v,
+              "onDeleteToString round-trips through parseOnDelete");
+    }
+
+    // A persisted record carrying a bad value falls back to Restrict (the safe
+    // direction: over-restricting refuses deletes, it never performs an
+    // unintended destructive one) rather than being dropped, which would remove
+    // protection entirely.
+    Fixture f;
+    Document d;
+    d.id = "default:bad_od";
+    d.set_data(nlohmann::json{
+        {"name", "default:bad_od"},
+        {"child", "default:executions"},
+        {"child_field", "workflowId"},
+        {"parent", "default:workflows"},
+        {"on_delete", "Cascade"}});
+    f.store.createCollection(RelationManager::SYSTEM_COLLECTION, CollectionOptions{});
+    f.store.upsert(RelationManager::SYSTEM_COLLECTION, d);
+
+    RelationManager rm(f.store);
+    rm.loadFromStore();
+    auto got = rm.getRelation("default:bad_od");
+    check(got.has_value(),
+          "a persisted record with a bad on_delete is still LOADED - dropping it "
+          "would silently remove protection");
+    check(got && got->onDelete == OnDelete::Restrict,
+          "and it falls back to restrict, the non-destructive direction");
+}
+
 int main() {
     std::cout << "=== test_relation_manager ===\n";
     test_relations_are_project_scoped_and_survive_reload();
     test_cross_project_relation_is_refused();
     test_more_than_one_page_of_relations_loads();
     test_load_skips_a_cross_project_relation_written_by_hand();
+    test_bare_and_qualified_names_are_the_same_relation();
+    test_load_rekeys_a_legacy_bare_named_record();
+    test_on_delete_is_validated_not_coerced();
     std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
     return g_fail == 0 ? 0 : 1;
 }