feat(security): principal identity + access policy engine (enforcement WIP)

NOT a release. Security cannot be enabled in this state - see the guard below.

Principal identity (spec stage v2.7.0, complete):
- auth.keys[] entries may now be {"name","key"}; the name becomes the PRINCIPAL
  that policy is written against. Bare strings still authenticate and map to the
  reserved principal `unnamed`, so existing configs keep working.
- BearerAuthProcessor stamps the matched key's name into the gRPC AuthContext.
  That is the only channel from an AuthMetadataProcessor to a handler, which is
  why the `context` argument ignored since v2.4 is finally used.
- auth::principalOf() resolves it back, returning the grantable `anonymous` when
  no processor ran - the default 127.0.0.1 listener has no auth and the CLI
  rides on it, so local access should be an explicit grant, not a hole.
- Startup refuses a key named `anonymous`/`unnamed` (impersonation) or a
  duplicate name (a principal must identify one caller or policy is ambiguous).

Policy engine (spec stage v2.8.0, engine complete, enforcement partial):
- PolicyManager over the `_policies` collection, id "<project>:<principal>",
  plus "<project>:__security__" for the per-project enable flag and mode.
  Durable for free: `_policies` is an ordinary collection, so it is WAL'd and
  snapshotted like everything else - no new persistence machinery, and no new
  WAL op type as putting this in CollectionOptions would have needed.
- Security is OFF per project by default; with it off authorize() allows
  everything behind one relaxed atomic load, so upgrades are inert.
- Once ON: deny-by-default, "*" fallback with exact-match precedence, row
  predicates AND-merged like a view's `where`, column masks reusing
  applyProjection, and system collections admin-only (read access to
  `_policies` would expose the access model, write access would be escalation).
- Lockout is structurally impossible: enabling is refused without an admin
  policy, and the last admin of a secured project cannot be removed.
- Audit mode evaluates honestly, logs what it would deny, and allows - the
  migration path for arming a live project.
- loadFromStore pages explicitly: Query::limit defaults to 100 and limit=0
  returns nothing, so a naive read would silently load a partial policy set and
  under-enforce. ViewManager has the same latent truncation past 100 views.

Enforcement is NOT complete, and that is enforced rather than documented:
kEnforcementCoverageComplete is false, so setSecurity(enabled=true) is refused.
A project enforcing on some paths and silently allowing on others reports
protection it does not have - strictly worse than security being off. Gated so
far: Get, Count. The header lists every remaining call site, plus the
outstanding mask application, masked-field write rejection, and the management
surface. A test asserts the guard itself.

Tests: tests/test_policy_manager.cpp, 56 assertions. ctest 17/17, e2e 29/29
and TLS/auth 4/4 green.

service/CMakeLists.txt

@@ -69,6 +69,8 @@ set(DATABASE_SERVICE_SOURCES
     src/storage/project_store.cpp
     src/tls/cert_generator.cpp
     src/auth/auth_interceptor.cpp
+    src/auth/principal.cpp
+    src/security/policy_manager.cpp
 )
 
 # Create executable

service/src/auth/auth_interceptor.cpp

@@ -2,6 +2,8 @@
 
 #include "auth/auth_interceptor.hpp"
 
+#include "auth/principal.hpp"
+
 #include <cstring>
 #include <string_view>
 
@@ -32,7 +34,7 @@ bool starts_with_bearer(std::string_view s) {
 
 grpc::Status BearerAuthProcessor::Process(
     const InputMetadata& auth_metadata,
-    grpc::AuthContext* /*context*/,
+    grpc::AuthContext* context,
     OutputMetadata* consumed_auth_metadata,
     OutputMetadata* /*response_metadata*/) {
 
@@ -51,12 +53,24 @@ grpc::Status BearerAuthProcessor::Process(
     std::string_view tok = value.substr(7);
 
     for (const auto& k : keys_) {
-        if (constant_time_eq(tok, k)) {
+        if (constant_time_eq(tok, k.key)) {
             // Mark the metadata as consumed so gRPC doesn't surface it
             // to handler code.
             consumed_auth_metadata->insert(
                 std::make_pair(std::string("authorization"),
                                 std::string(value)));
+
+            // v2.7.0 — stamp the principal. This is the ONLY channel from an
+            // AuthMetadataProcessor to a handler, so policy enforcement
+            // downstream depends on it. Empty name should be impossible
+            // (config normalises bare keys to `unnamed`) but fall back rather
+            // than stamping an empty identity.
+            if (context) {
+                const std::string principal =
+                    k.name.empty() ? std::string(kPrincipalUnnamed) : k.name;
+                context->AddProperty(kPrincipalProperty, principal);
+                context->SetPeerIdentityPropertyName(kPrincipalProperty);
+            }
             return grpc::Status::OK;
         }
     }

service/src/auth/auth_interceptor.hpp

@@ -7,6 +7,12 @@
 // gRPC runtime handles rejection without each handler having to
 // check. Listeners with auth.required=false don't register a
 // processor at all.
+//
+// v2.7.0 — keys are named, and the matched key's name is stamped into the
+// AuthContext as the calling PRINCIPAL. That is the only channel gRPC offers
+// between an AuthMetadataProcessor and a handler, which is why the `context`
+// argument (ignored until now) is finally used. Handlers read it back via
+// auth::principalOf(). See auth/principal.hpp.
 
 #pragma once
 
@@ -21,7 +27,12 @@ namespace smartbotic::database::auth {
 
 class BearerAuthProcessor : public grpc::AuthMetadataProcessor {
 public:
-    explicit BearerAuthProcessor(std::vector<std::string> keys)
+    struct NamedKey {
+        std::string name;  // principal stamped on a match
+        std::string key;   // bearer token
+    };
+
+    explicit BearerAuthProcessor(std::vector<NamedKey> keys)
         : keys_(std::move(keys)) {}
 
     grpc::Status Process(const InputMetadata& auth_metadata,
@@ -34,7 +45,7 @@ public:
     bool IsBlocking() const override { return false; }
 
 private:
-    std::vector<std::string> keys_;
+    std::vector<NamedKey> keys_;
 };
 
 }  // namespace smartbotic::database::auth

service/src/auth/principal.cpp

@@ -0,0 +1,25 @@
+// v2.7.0 — caller identity. See principal.hpp.
+
+#include "auth/principal.hpp"
+
+#include <grpcpp/grpcpp.h>
+
+namespace smartbotic::database::auth {
+
+std::string principalOf(const grpc::ServerContext* context) {
+    if (!context) return kPrincipalAnonymous;
+
+    auto auth_context = context->auth_context();
+    if (!auth_context) return kPrincipalAnonymous;
+
+    // Listeners without auth register no processor, so the property is absent.
+    // That is the `anonymous` case, not an error.
+    auto values = auth_context->FindPropertyValues(kPrincipalProperty);
+    if (values.empty()) return kPrincipalAnonymous;
+
+    std::string name(values[0].data(), values[0].size());
+    if (name.empty()) return kPrincipalAnonymous;
+    return name;
+}
+
+}  // namespace smartbotic::database::auth

service/src/auth/principal.hpp

@@ -0,0 +1,52 @@
+// v2.7.0 — caller identity.
+//
+// Until now the server knew only "the token was valid". It attached nothing to
+// the call, so every key holder was indistinguishable and access control was
+// impossible to express. This is the identity layer that per-project row- and
+// column-level policy is written against.
+//
+// How a principal is decided, in order:
+//   1. the listener has auth and the token matched a NAMED key  -> that name
+//   2. the listener has auth and the token matched a BARE key    -> `unnamed`
+//   3. the listener has no auth processor at all                 -> `anonymous`
+//
+// Case 3 is not an error. The default 127.0.0.1 plaintext listener has no auth,
+// and `smartbotic-db-cli` plus operator tooling ride on it. `anonymous` is a
+// real, grantable principal so local access is allowed by an explicit policy
+// rather than by an implicit hole. See
+// docs/superpowers/specs/2026-08-08-rls-cls-design.md.
+//
+// Transport: BearerAuthProcessor stores the name as an auth-context property.
+// gRPC gives no way to pass state from an AuthMetadataProcessor to a handler
+// other than the AuthContext, which is why the processor's `context` argument
+// (ignored since v2.4) is now used.
+
+#pragma once
+
+#include <string>
+#include <string_view>
+
+namespace grpc {
+class ServerContext;
+}
+
+namespace smartbotic::database::auth {
+
+// Auth-context property carrying the principal name.
+inline constexpr const char* kPrincipalProperty = "sbdb_principal";
+
+// Reserved principal names. Rejected as key names in config, because a key
+// called "anonymous" would otherwise be indistinguishable from an unauthed
+// caller in a policy.
+inline constexpr const char* kPrincipalAnonymous = "anonymous";
+inline constexpr const char* kPrincipalUnnamed   = "unnamed";
+
+inline bool isReservedPrincipal(std::string_view name) {
+    return name == kPrincipalAnonymous || name == kPrincipalUnnamed;
+}
+
+// Resolve the calling principal. Never throws and never returns empty: a
+// request with no identity is `anonymous`, which is a decision, not a failure.
+std::string principalOf(const grpc::ServerContext* context);
+
+}  // namespace smartbotic::database::auth

service/src/database_grpc_impl.cpp

@@ -86,7 +86,8 @@ DatabaseGrpcImpl::DatabaseGrpcImpl(
     FileManager& files,
     EncryptionManager& encryption,
     ViewManager& view_manager,
-    CollectionConfigManager& config_manager
+    CollectionConfigManager& config_manager,
+    PolicyManager& policy_manager
 ) : service_(service)
   , store_(store)
   , persistence_(persistence)
@@ -95,13 +96,102 @@ DatabaseGrpcImpl::DatabaseGrpcImpl(
   , encryption_(encryption)
   , view_manager_(view_manager)
   , config_manager_(config_manager)
+  , policy_manager_(policy_manager)
 {
 }
 
 // ===== Document Operations =====
 
+
+// -------------------------------------------------------------------------
+// v2.8.0 — access gate. See database_grpc_impl.hpp for why this is one
+// function rather than a check per handler.
+// -------------------------------------------------------------------------
+
+grpc::Status DatabaseGrpcImpl::gate(const grpc::ServerContext* context,
+                                     const std::string& qualified,
+                                     smartbotic::database::Access access,
+                                     smartbotic::database::Decision& out) const {
+    // Fast path: no project has security enabled, which is the default and the
+    // whole pre-2.8.0 world. One relaxed atomic load.
+    if (!policy_manager_.anyProjectSecured()) {
+        out = smartbotic::database::Decision{};
+        return grpc::Status::OK;
+    }
+
+    smartbotic::database::ProjectCollection pc;
+    try {
+        pc = smartbotic::database::parseProjectCollection(qualified);
+    } catch (const std::exception&) {
+        // An unparseable name cannot be authorised. Fail closed.
+        return grpc::Status(grpc::StatusCode::PERMISSION_DENIED,
+                            "access denied");
+    }
+
+    const std::string principal = smartbotic::database::auth::principalOf(context);
+    out = policy_manager_.authorize(pc.project, principal, pc.collection, access);
+    if (!out.allowed) {
+        // Deliberately terse and identical for every denial reason. A response
+        // that distinguished "no policy" from "collection not granted" would
+        // let a caller map the access model by probing.
+        spdlog::warn("policy: DENIED principal '{}' {} on '{}' ({})",
+                     principal, access == smartbotic::database::Access::Read ? "read" : "write",
+                     qualified, out.reason);
+        return grpc::Status(grpc::StatusCode::PERMISSION_DENIED, "access denied");
+    }
+    return grpc::Status::OK;
+}
+
+grpc::Status DatabaseGrpcImpl::gateFile(const grpc::ServerContext* context,
+                                         const std::string& project,
+                                         const std::string& fileType,
+                                         smartbotic::database::Access access,
+                                         smartbotic::database::Decision& out) const {
+    if (!policy_manager_.anyProjectSecured()) {
+        out = smartbotic::database::Decision{};
+        return grpc::Status::OK;
+    }
+    const std::string proj = project.empty() ? "default" : project;
+    const std::string principal = smartbotic::database::auth::principalOf(context);
+    out = policy_manager_.authorizeFile(proj, principal, fileType, access);
+    if (!out.allowed) {
+        spdlog::warn("policy: DENIED principal '{}' {} on file type '{}' in '{}' ({})",
+                     principal, access == smartbotic::database::Access::Read ? "read" : "write",
+                     fileType, proj, out.reason);
+        return grpc::Status(grpc::StatusCode::PERMISSION_DENIED, "access denied");
+    }
+    return grpc::Status::OK;
+}
+
+void DatabaseGrpcImpl::applyMask(smartbotic::database::Document& doc,
+                                  const std::vector<std::string>& mask) {
+    if (mask.empty()) return;
+    // Reuse the view machinery: a column mask is exactly a view `exclude` list,
+    // same dot-notation paths, same semantics.
+    nlohmann::json data = doc.data();
+    data = applyProjection(data, /*include=*/{}, /*exclude=*/mask);
+    doc.set_data(data);
+}
+
+grpc::Status DatabaseGrpcImpl::rejectMaskedFilters(
+    const std::vector<smartbotic::database::Filter>& filters,
+    const std::vector<std::string>& mask) {
+    if (mask.empty()) return grpc::Status::OK;
+    for (const auto& f : filters) {
+        for (const auto& m : mask) {
+            // Exact path, or the filter reaching into a masked subtree.
+            if (f.field == m || f.field.rfind(m + ".", 0) == 0) {
+                return grpc::Status(
+                    grpc::StatusCode::INVALID_ARGUMENT,
+                    "filtering on field '" + f.field + "' is not permitted");
+            }
+        }
+    }
+    return grpc::Status::OK;
+}
+
 grpc::Status DatabaseGrpcImpl::Insert(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::InsertRequest* request,
     pb::InsertResponse* response
 ) {
@@ -159,7 +249,7 @@ grpc::Status DatabaseGrpcImpl::Insert(
 }
 
 grpc::Status DatabaseGrpcImpl::Get(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetRequest* request,
     pb::GetResponse* response
 ) {
@@ -171,6 +261,14 @@ grpc::Status DatabaseGrpcImpl::Get(
         targetCollection = view->collection;
     }
 
+    // v2.8.0 — access gate. Placed AFTER view resolution so policy applies to
+    // the underlying collection: a view must not become a way around a rule.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, targetCollection, smartbotic::database::Access::Read, dec);
+        !st.ok()) {
+        return st;
+    }
+
     // v2.0 Stage 4 — LMDB-first read. Gated on mirror_healthy_ + zero
     // drift + non-system collection + doc_store_ available. The
     // defensive MemoryStore-on-LMDB-absent fallback was removed once the
@@ -239,7 +337,7 @@ grpc::Status DatabaseGrpcImpl::Get(
 }
 
 grpc::Status DatabaseGrpcImpl::Update(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::UpdateRequest* request,
     pb::UpdateResponse* response
 ) {
@@ -322,7 +420,7 @@ grpc::Status DatabaseGrpcImpl::Update(
 }
 
 grpc::Status DatabaseGrpcImpl::PatchDocument(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::PatchDocumentRequest* request,
     pb::PatchDocumentResponse* response
 ) {
@@ -456,7 +554,7 @@ grpc::Status DatabaseGrpcImpl::Upsert(
 }
 
 grpc::Status DatabaseGrpcImpl::Delete(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::DeleteRequest* request,
     pb::DeleteResponse* response
 ) {
@@ -488,7 +586,7 @@ grpc::Status DatabaseGrpcImpl::Delete(
 }
 
 grpc::Status DatabaseGrpcImpl::Exists(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::ExistsRequest* request,
     pb::ExistsResponse* response
 ) {
@@ -530,7 +628,7 @@ grpc::Status DatabaseGrpcImpl::Exists(
 // ===== Version History Operations =====
 
 grpc::Status DatabaseGrpcImpl::GetVersionHistory(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetVersionHistoryRequest* request,
     pb::GetVersionHistoryResponse* response
 ) {
@@ -726,7 +824,7 @@ grpc::Status DatabaseGrpcImpl::BatchDelete(
 // ===== Query Operations =====
 
 grpc::Status DatabaseGrpcImpl::Find(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::FindRequest* request,
     pb::FindResponse* response
 ) {
@@ -818,7 +916,7 @@ grpc::Status DatabaseGrpcImpl::Find(
 }
 
 grpc::Status DatabaseGrpcImpl::Count(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::CountRequest* request,
     pb::CountResponse* response
 ) {
@@ -839,6 +937,17 @@ grpc::Status DatabaseGrpcImpl::Count(
         filters.push_back(std::move(f));
     }
 
+    // v2.8.0 — access gate. A denied principal must not learn the size of a
+    // collection it cannot read; the count itself is information.
+    smartbotic::database::Decision dec;
+    if (auto st = gate(context, targetCollection, smartbotic::database::Access::Read, dec);
+        !st.ok()) {
+        return st;
+    }
+    if (auto st = rejectMaskedFilters(filters, dec.mask); !st.ok()) return st;
+    // Row predicate narrows the count, exactly as it narrows a Find.
+    for (const auto& f : dec.row) filters.push_back(f);
+
     // v2.4.4 — LMDB-first Count, same gate as Exists/Get/Find.
     //
     // Until now Count went to MemoryStore unconditionally while every other
@@ -885,7 +994,7 @@ grpc::Status DatabaseGrpcImpl::Count(
 }
 
 grpc::Status DatabaseGrpcImpl::SimilaritySearch(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::SimilaritySearchRequest* request,
     pb::SimilaritySearchResponse* response
 ) {
@@ -1188,7 +1297,7 @@ grpc::Status DatabaseGrpcImpl::ListCollections(
 }
 
 grpc::Status DatabaseGrpcImpl::GetCollectionInfo(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetCollectionInfoRequest* request,
     pb::GetCollectionInfoResponse* response
 ) {
@@ -1285,7 +1394,7 @@ grpc::Status DatabaseGrpcImpl::DropProject(
 // ===== File Operations =====
 
 grpc::Status DatabaseGrpcImpl::UploadFile(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     grpc::ServerReader<pb::FileChunk>* reader,
     pb::UploadFileResponse* response
 ) {
@@ -1396,7 +1505,7 @@ grpc::Status DatabaseGrpcImpl::DownloadFile(
 }
 
 grpc::Status DatabaseGrpcImpl::DeleteFile(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::DeleteFileRequest* request,
     pb::DeleteFileResponse* response
 ) {
@@ -1413,7 +1522,7 @@ grpc::Status DatabaseGrpcImpl::DeleteFile(
 }
 
 grpc::Status DatabaseGrpcImpl::GetFileInfo(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::GetFileInfoRequest* request,
     pb::FileInfo* response
 ) {
@@ -1441,7 +1550,7 @@ grpc::Status DatabaseGrpcImpl::GetFileInfo(
 }
 
 grpc::Status DatabaseGrpcImpl::ListFiles(
-    grpc::ServerContext* /*context*/,
+    grpc::ServerContext* context,
     const pb::ListFilesRequest* request,
     pb::ListFilesResponse* response
 ) {

service/src/database_grpc_impl.hpp

@@ -8,6 +8,8 @@
 #include "replication/replication_manager.hpp"
 #include "views/view_manager.hpp"
 #include "config/collection_config_manager.hpp"
+#include "security/policy_manager.hpp"
+#include "auth/principal.hpp"
 
 #include <database.grpc.pb.h>
 
@@ -39,7 +41,8 @@ public:
         FileManager& files,
         EncryptionManager& encryption,
         ViewManager& view_manager,
-        CollectionConfigManager& config_manager
+        CollectionConfigManager& config_manager,
+        PolicyManager& policy_manager
     );
 
     ~DatabaseGrpcImpl() override = default;
@@ -370,6 +373,39 @@ private:
     EncryptionManager& encryption_;
     ViewManager& view_manager_;
     CollectionConfigManager& config_manager_;
+    PolicyManager& policy_manager_;
+
+    // v2.8.0 — the single access gate every handler goes through.
+    //
+    // `qualified` is the "<project>:<collection>" form the handler already has.
+    // On allow, `out` carries the column mask and row predicate to apply. On
+    // deny, the returned Status is PERMISSION_DENIED and the handler must
+    // return it untouched.
+    //
+    // Deliberately ONE function rather than inline checks: with ~16 call sites,
+    // a per-handler check is a place to forget, and forgetting fails open.
+    grpc::Status gate(const grpc::ServerContext* context,
+                      const std::string& qualified,
+                      smartbotic::database::Access access,
+                      smartbotic::database::Decision& out) const;
+
+    // Same for the file RPCs, which key on file type rather than collection.
+    grpc::Status gateFile(const grpc::ServerContext* context,
+                          const std::string& project,
+                          const std::string& fileType,
+                          smartbotic::database::Access access,
+                          smartbotic::database::Decision& out) const;
+
+    // Strip masked paths from a document's data in place. No-op on empty mask.
+    static void applyMask(smartbotic::database::Document& doc,
+                          const std::vector<std::string>& mask);
+
+    // Reject a filter that names a masked path. Allowing it would turn the mask
+    // into an inference channel: `salary > 100000` leaks a value the caller is
+    // not permitted to read.
+    static grpc::Status rejectMaskedFilters(
+        const std::vector<smartbotic::database::Filter>& filters,
+        const std::vector<std::string>& mask);
     std::chrono::steady_clock::time_point startTime_ = std::chrono::steady_clock::now();
 
     // v1.6.2 — per-RPC-type streaming concurrency limits (set by DatabaseService

service/src/database_service.cpp

@@ -5,11 +5,13 @@
 #include "storage/document_store_lmdb.hpp"
 #include "storage/dual_write_mirror.hpp"
 #include "auth/auth_interceptor.hpp"
+#include "auth/principal.hpp"
 #include "tls/cert_generator.hpp"
 
 #include <fstream>
 #include <sstream>
 #include <map>
+#include <set>
 #include "storage/lmdb_env.hpp"
 #include "storage/migrate_v1_to_v2.hpp"
 #include "storage/project_store.hpp"
@@ -192,6 +194,7 @@ bool DatabaseService::initialize() {
         // Must happen AFTER persistence recovery and BEFORE migrations so any
         // migration-created documents are stamped with the correct precision.
         config_manager_->loadFromStore();
+        policy_manager_->loadFromStore();
 
         // Run migrations if enabled
         if (config_.migrations.enabled && !config_.migrations.directory.empty()) {
@@ -632,7 +635,31 @@ DatabaseService::Config DatabaseService::parseConfig(const nlohmann::json& json)
                     lc.auth.required = a.value("required", false);
                     if (a.contains("keys") && a["keys"].is_array()) {
                         for (const auto& k : a["keys"]) {
-                            if (k.is_string()) lc.auth.keys.push_back(k.get<std::string>());
+                            // v2.7.0 — a key is either a bare token (v2.4-v2.6
+                            // shape) or {"name","key"}. The name becomes the
+                            // principal that policy is written against; a bare
+                            // token maps to the reserved `unnamed`, so old
+                            // configs keep authenticating but cannot be told
+                            // apart in a policy until they are named.
+                            if (k.is_string()) {
+                                lc.auth.keys.push_back(
+                                    {smartbotic::database::auth::kPrincipalUnnamed,
+                                     k.get<std::string>()});
+                            } else if (k.is_object()) {
+                                const std::string name = k.value("name", std::string{});
+                                const std::string key  = k.value("key", std::string{});
+                                if (key.empty()) {
+                                    spdlog::warn("auth: listener {}:{} has a key entry "
+                                                 "with no `key` value - skipped",
+                                                 lc.bind, lc.port);
+                                    continue;
+                                }
+                                lc.auth.keys.push_back(
+                                    {name.empty()
+                                         ? std::string(smartbotic::database::auth::kPrincipalUnnamed)
+                                         : name,
+                                     key});
+                            }
                         }
                     }
                 }
@@ -886,6 +913,9 @@ void DatabaseService::setupComponents() {
     // Create per-collection config manager and attach it to the store so the
     // write paths route document timestamp stamps through it.
     config_manager_ = std::make_unique<CollectionConfigManager>(*store_);
+    // v2.8.0 — access policy. Constructed here; its cache is loaded after
+    // recovery alongside the view and collection-config caches.
+    policy_manager_ = std::make_unique<PolicyManager>(*store_);
     store_->setConfigManager(config_manager_.get());
 
     // v1.9.0 — disk-resident version history. Replaces the in-heap
@@ -1062,7 +1092,8 @@ void DatabaseService::setupComponents() {
 
     // Create gRPC implementations
     storageImpl_ = std::make_unique<DatabaseGrpcImpl>(
-        *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_
+        *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_,
+        *policy_manager_
     );
     // v1.6.2 — wire the streaming-RPC concurrency limits from GrpcConfig.
     storageImpl_->setStreamLimits(
@@ -1266,6 +1297,29 @@ void DatabaseService::startGrpcServer() {
                 throw std::runtime_error(
                     "Listener " + addr + ": auth.required=true but auth.keys is empty");
             }
+            // v2.7.0 — a key name is a principal, so it must be unambiguous.
+            // Refuse at startup rather than resolving a policy against a name
+            // that means two different callers.
+            {
+                std::set<std::string> seen;
+                for (const auto& k : listener.auth.keys) {
+                    if (k.name != smartbotic::database::auth::kPrincipalUnnamed &&
+                        smartbotic::database::auth::isReservedPrincipal(k.name)) {
+                        throw std::runtime_error(
+                            "Listener " + addr + ": auth key name '" + k.name +
+                            "' is reserved. `anonymous` denotes an unauthenticated "
+                            "caller and `unnamed` denotes a legacy bare key; a real "
+                            "key must not be able to impersonate either in a policy.");
+                    }
+                    if (!seen.insert(k.name).second &&
+                        k.name != smartbotic::database::auth::kPrincipalUnnamed) {
+                        throw std::runtime_error(
+                            "Listener " + addr + ": duplicate auth key name '" +
+                            k.name + "'. A principal must identify one caller, "
+                            "otherwise a policy written against it is ambiguous.");
+                    }
+                }
+            }
             if (!listener.tls.enabled) {
                 spdlog::warn(
                     "Listener {}: auth.required=true but tls.enabled=false — "
@@ -1273,9 +1327,12 @@ void DatabaseService::startGrpcServer() {
                     "Consider enabling TLS for any non-loopback listener.",
                     addr);
             }
+            std::vector<smartbotic::database::auth::BearerAuthProcessor::NamedKey> nk;
+            nk.reserve(listener.auth.keys.size());
+            for (const auto& k : listener.auth.keys) nk.push_back({k.name, k.key});
             creds->SetAuthMetadataProcessor(
                 std::make_shared<smartbotic::database::auth::BearerAuthProcessor>(
-                    listener.auth.keys));
+                    std::move(nk)));
         }
         builder.AddListeningPort(addr, creds);
 

service/src/database_service.hpp

@@ -12,6 +12,7 @@
 #include "migrations/migration_runner.hpp"
 #include "views/view_manager.hpp"
 #include "config/collection_config_manager.hpp"
+#include "security/policy_manager.hpp"
 
 // v2.0 storage substrate (Stage 4) — DocumentStore + LmdbEnv live alongside
 // the v1.x MemoryStore during the Phase C transition. Stage 4 opens both;
@@ -67,12 +68,28 @@ public:
 
         struct AuthConfig {
             bool required = false;
+
+            // v2.7.0 — a key now carries a name, which becomes the
+            // PRINCIPAL that access policy is written against.
+            //
+            // Config accepts either form:
+            //   "keys": [ {"name": "shadowman", "key": "<base64>"} ]
+            //   "keys": [ "<base64>" ]            // v2.4-v2.6 shape
+            //
+            // A bare string still authenticates and maps to the reserved
+            // principal `unnamed`, so existing deployments keep working; they
+            // simply cannot be told apart in a policy until they are named.
+            struct NamedKey {
+                std::string name;   // principal; "unnamed" when config gave a bare string
+                std::string key;    // the bearer token itself
+            };
+
             // Bearer tokens accepted by this listener. Constant-time
             // compared against the value of the `authorization` gRPC
             // metadata header. Rotation = add new key, distribute,
             // remove old key. Empty list with required=true is a config
             // error caught at startup.
-            std::vector<std::string> keys;
+            std::vector<NamedKey> keys;
         } auth;
     };
 
@@ -214,6 +231,7 @@ public:
      * Access the view manager (for RPC handlers, migrations, etc.).
      */
     ViewManager& viewManager() { return *view_manager_; }
+    PolicyManager& policyManager() { return *policy_manager_; }
 
     /** Current read-only state (atomic, lock-free read). */
     bool isReadOnly() const { return read_only_.load(std::memory_order_acquire); }
@@ -328,6 +346,9 @@ private:
     std::unique_ptr<ReplicationManager> replication_;
     std::unique_ptr<ViewManager> view_manager_;
     std::unique_ptr<CollectionConfigManager> config_manager_;
+    // v2.8.0 — per-project access policy. Owned here so its cache is loaded
+    // once, after recovery, alongside ViewManager and CollectionConfigManager.
+    std::unique_ptr<PolicyManager> policy_manager_;
 
     // gRPC
     std::unique_ptr<DatabaseGrpcImpl> storageImpl_;

service/src/security/policy_manager.cpp

@@ -0,0 +1,438 @@
+// v2.8.0 — per-project access policy. See policy_manager.hpp.
+
+#include "security/policy_manager.hpp"
+
+#include <nlohmann/json.hpp>
+#include <spdlog/spdlog.h>
+
+#include "memory_store.hpp"
+
+namespace smartbotic::database {
+
+namespace {
+
+std::string keyOf(const std::string& project, const std::string& principal) {
+    return project + ":" + principal;
+}
+
+nlohmann::json ruleToJson(const PolicyRule& r) {
+    nlohmann::json j;
+    j["read"] = r.read;
+    j["write"] = r.write;
+    j["mask"] = r.mask;
+    nlohmann::json rows = nlohmann::json::array();
+    for (const auto& f : r.row) {
+        rows.push_back({{"field", f.field},
+                        {"op", static_cast<int>(f.op)},
+                        {"value", f.value}});
+    }
+    j["row"] = rows;
+    return j;
+}
+
+PolicyRule ruleFromJson(const nlohmann::json& j) {
+    PolicyRule r;
+    r.read = j.value("read", false);
+    r.write = j.value("write", false);
+    if (j.contains("mask") && j["mask"].is_array()) {
+        for (const auto& m : j["mask"]) {
+            if (m.is_string()) r.mask.push_back(m.get<std::string>());
+        }
+    }
+    if (j.contains("row") && j["row"].is_array()) {
+        for (const auto& f : j["row"]) {
+            if (!f.is_object()) continue;
+            Filter flt;
+            flt.field = f.value("field", "");
+            flt.op = static_cast<FilterOp>(f.value("op", 0));
+            if (f.contains("value")) flt.value = f["value"];
+            if (!flt.field.empty()) r.row.push_back(std::move(flt));
+        }
+    }
+    return r;
+}
+
+nlohmann::json policyToJson(const Policy& p) {
+    nlohmann::json j;
+    j["principal"] = p.principal;
+    j["admin"] = p.admin;
+    nlohmann::json colls = nlohmann::json::object();
+    for (const auto& [k, v] : p.collections) colls[k] = ruleToJson(v);
+    j["collections"] = colls;
+    nlohmann::json files = nlohmann::json::object();
+    for (const auto& [k, v] : p.files) files[k] = ruleToJson(v);
+    j["files"] = files;
+    return j;
+}
+
+Policy policyFromJson(const nlohmann::json& j) {
+    Policy p;
+    p.principal = j.value("principal", "");
+    p.admin = j.value("admin", false);
+    if (j.contains("collections") && j["collections"].is_object()) {
+        for (const auto& [k, v] : j["collections"].items()) {
+            p.collections[k] = ruleFromJson(v);
+        }
+    }
+    if (j.contains("files") && j["files"].is_object()) {
+        for (const auto& [k, v] : j["files"].items()) {
+            p.files[k] = ruleFromJson(v);
+        }
+    }
+    return p;
+}
+
+}  // namespace
+
+PolicyManager::PolicyManager(MemoryStore& store) : store_(store) {}
+
+void PolicyManager::loadFromStore() {
+    std::unique_lock lock(mutex_);
+    policies_.clear();
+    security_.clear();
+
+    CollectionOptions opts;
+    opts.autoCreateId = false;
+    store_.createCollection(SYSTEM_COLLECTION, opts);
+
+    // Query::limit defaults to 100 and limit=0 returns NOTHING, not everything
+    // (`end = min(offset + limit, size)`), so paging is mandatory here. Loading
+    // a partial policy set would silently under-enforce, which is the worst
+    // possible failure for this component. Note ViewManager has the same latent
+    // truncation on installs with more than 100 views.
+    std::vector<Document> records;
+    {
+        constexpr uint32_t kPage = 500;
+        uint32_t offset = 0;
+        while (true) {
+            Query q;
+            q.limit = kPage;
+            q.offset = offset;
+            QueryResult page;
+            try {
+                page = store_.find(SYSTEM_COLLECTION, q);
+            } catch (const std::exception& e) {
+                spdlog::warn("PolicyManager: could not read {}: {}",
+                             SYSTEM_COLLECTION, e.what());
+                return;
+            }
+            for (auto& d : page.documents) records.push_back(std::move(d));
+            if (page.documents.size() < kPage) break;
+            offset += kPage;
+        }
+    }
+
+    uint32_t policies = 0, projects = 0;
+    for (const auto& doc : records) {
+        // id is "<project>:<principal>" or "<project>:__security__"
+        const auto sep = doc.id.find(':');
+        if (sep == std::string::npos) {
+            spdlog::warn("PolicyManager: skipping unqualified policy id '{}'", doc.id);
+            continue;
+        }
+        const std::string project = doc.id.substr(0, sep);
+        const std::string tail = doc.id.substr(sep + 1);
+
+        try {
+            const auto data = doc.data();
+            if (tail == SECURITY_DOC) {
+                ProjectSecurity sec;
+                sec.enabled = data.value("enabled", false);
+                sec.mode = data.value("mode", std::string("enforce")) == "audit"
+                               ? SecurityMode::Audit
+                               : SecurityMode::Enforce;
+                security_[project] = sec;
+                ++projects;
+            } else {
+                Policy p = policyFromJson(data);
+                if (p.principal.empty()) p.principal = tail;
+                policies_[doc.id] = std::move(p);
+                ++policies;
+            }
+        } catch (const std::exception& e) {
+            spdlog::warn("PolicyManager: skipping malformed record '{}': {}",
+                         doc.id, e.what());
+        }
+    }
+
+    recountSecured();
+    if (policies || projects) {
+        spdlog::info("PolicyManager: loaded {} policy record(s) across {} "
+                     "configured project(s); {} project(s) have security enabled",
+                     policies, projects,
+                     secured_count_.load(std::memory_order_relaxed));
+    }
+}
+
+void PolicyManager::recountSecured() {
+    uint32_t n = 0;
+    for (const auto& [_, sec] : security_) {
+        if (sec.enabled) ++n;
+    }
+    secured_count_.store(n, std::memory_order_relaxed);
+}
+
+ProjectSecurity PolicyManager::securityOf(const std::string& project) const {
+    std::shared_lock lock(mutex_);
+    auto it = security_.find(project);
+    if (it == security_.end()) return {};  // disabled
+    return it->second;
+}
+
+Decision PolicyManager::evaluate(const std::string& project,
+                                  const std::string& principal,
+                                  const std::string& target,
+                                  Access access,
+                                  bool is_file) const {
+    std::shared_lock lock(mutex_);
+
+    auto sec_it = security_.find(project);
+    const ProjectSecurity sec =
+        (sec_it == security_.end()) ? ProjectSecurity{} : sec_it->second;
+
+    // Security off: allow everything. This is the default and the entire
+    // pre-2.8.0 world, so it must cost almost nothing.
+    if (!sec.enabled) return Decision{};
+
+    Decision d;
+    auto deny = [&](const std::string& why) {
+        d.allowed = false;
+        d.reason = why;
+        // Audit mode: evaluate honestly, log, then allow anyway. The caller is
+        // told nothing; the operator sees exactly what enforce mode would do.
+        if (sec.mode == SecurityMode::Audit) {
+            d.allowed = true;
+            d.audited_denial = true;
+            spdlog::warn("policy AUDIT: would deny principal '{}' {} on {} '{}' "
+                         "in project '{}' ({}). Not denied - project is in audit "
+                         "mode.",
+                         principal,
+                         access == Access::Read ? "read" : "write",
+                         is_file ? "file type" : "collection",
+                         target, project, why);
+        }
+        return d;
+    };
+
+    // System collections are never reachable by a non-admin. `_policies` in
+    // particular: read access to it would expose the whole access model, and
+    // write access would be privilege escalation.
+    const bool is_system = !is_file && !target.empty() && target.front() == '_';
+
+    auto pol_it = policies_.find(keyOf(project, principal));
+    if (pol_it == policies_.end()) {
+        return deny("no policy for this principal (project is deny-by-default)");
+    }
+    const Policy& p = pol_it->second;
+
+    if (p.admin) return d;  // admin: unrestricted within the project
+    if (is_system) return deny("system collections are admin-only");
+
+    const auto& table = is_file ? p.files : p.collections;
+    auto rule_it = table.find(target);
+    if (rule_it == table.end()) {
+        rule_it = table.find("*");
+        if (rule_it == table.end()) {
+            return deny("no rule for this " +
+                        std::string(is_file ? "file type" : "collection"));
+        }
+    }
+    const PolicyRule& rule = rule_it->second;
+
+    const bool granted = (access == Access::Read) ? rule.read : rule.write;
+    if (!granted) {
+        return deny(std::string(access == Access::Read ? "read" : "write") +
+                    " not granted");
+    }
+
+    d.mask = rule.mask;
+    d.row = rule.row;
+    return d;
+}
+
+Decision PolicyManager::authorize(const std::string& project,
+                                   const std::string& principal,
+                                   const std::string& collection,
+                                   Access access) const {
+    return evaluate(project, principal, collection, access, /*is_file=*/false);
+}
+
+Decision PolicyManager::authorizeFile(const std::string& project,
+                                       const std::string& principal,
+                                       const std::string& fileType,
+                                       Access access) const {
+    return evaluate(project, principal, fileType, access, /*is_file=*/true);
+}
+
+bool PolicyManager::mayAdminister(const std::string& project,
+                                   const std::string& principal) const {
+    std::shared_lock lock(mutex_);
+    auto sec_it = security_.find(project);
+    // Security off: policy is freely editable, which is how the first policy
+    // gets created in the first place.
+    if (sec_it == security_.end() || !sec_it->second.enabled) return true;
+
+    auto it = policies_.find(keyOf(project, principal));
+    return it != policies_.end() && it->second.admin;
+}
+
+bool PolicyManager::setSecurity(const std::string& project,
+                                 const ProjectSecurity& sec,
+                                 std::string& errorOut) {
+    if (sec.enabled && !allow_enable_) {
+        // Refuse to arm a boundary that is not fully wired yet.
+        //
+        // The engine, its safety rules and its tests are done, but the gate is
+        // not yet called from every read/write handler. A project enabled in
+        // that state would enforce on some paths and silently allow on others,
+        // which is worse than no security at all: it reports protection that is
+        // not there. See kEnforcementCoverageComplete in policy_manager.hpp for
+        // the remaining call sites.
+        errorOut =
+            "refusing to enable security: enforcement is not yet wired into "
+            "every handler in this build, so a project would be protected on "
+            "some paths and open on others. Policies can be created and "
+            "inspected now; enabling is unlocked once coverage is complete.";
+        return false;
+    }
+    if (sec.enabled) {
+        // Lockout prevention. Refusing here is the whole safety mechanism:
+        // enabling deny-by-default with no admin would leave the project
+        // permanently unmanageable, including by the operator.
+        std::shared_lock lock(mutex_);
+        bool has_admin = false;
+        const std::string prefix = project + ":";
+        for (const auto& [key, pol] : policies_) {
+            if (key.rfind(prefix, 0) == 0 && pol.admin) {
+                has_admin = true;
+                break;
+            }
+        }
+        if (!has_admin) {
+            errorOut = "refusing to enable security for project '" + project +
+                       "': no policy in this project has admin=true, so enabling "
+                       "deny-by-default would lock everyone out permanently. "
+                       "Create an admin policy first.";
+            return false;
+        }
+    }
+
+    Document doc;
+    doc.id = keyOf(project, SECURITY_DOC);
+    doc.collection = SYSTEM_COLLECTION;
+    nlohmann::json data;
+    data["enabled"] = sec.enabled;
+    data["mode"] = sec.mode == SecurityMode::Audit ? "audit" : "enforce";
+    doc.set_data(data);
+
+    try {
+        store_.remove(SYSTEM_COLLECTION, doc.id);
+        store_.insert(SYSTEM_COLLECTION, doc);
+    } catch (const std::exception& e) {
+        errorOut = std::string("failed to persist security record: ") + e.what();
+        return false;
+    }
+
+    {
+        std::unique_lock lock(mutex_);
+        security_[project] = sec;
+        recountSecured();
+    }
+    spdlog::warn("policy: project '{}' security {} (mode={})", project,
+                 sec.enabled ? "ENABLED" : "disabled",
+                 sec.mode == SecurityMode::Audit ? "audit" : "enforce");
+    return true;
+}
+
+bool PolicyManager::setPolicy(const std::string& project, const Policy& policy,
+                               std::string& errorOut) {
+    if (policy.principal.empty()) {
+        errorOut = "policy principal must not be empty";
+        return false;
+    }
+    if (policy.principal == SECURITY_DOC) {
+        errorOut = "'" + std::string(SECURITY_DOC) + "' is reserved";
+        return false;
+    }
+
+    Document doc;
+    doc.id = keyOf(project, policy.principal);
+    doc.collection = SYSTEM_COLLECTION;
+    doc.set_data(policyToJson(policy));
+
+    try {
+        store_.remove(SYSTEM_COLLECTION, doc.id);
+        store_.insert(SYSTEM_COLLECTION, doc);
+    } catch (const std::exception& e) {
+        errorOut = std::string("failed to persist policy: ") + e.what();
+        return false;
+    }
+
+    {
+        std::unique_lock lock(mutex_);
+        policies_[doc.id] = policy;
+    }
+    spdlog::info("policy: set for principal '{}' in project '{}' (admin={})",
+                 policy.principal, project, policy.admin);
+    return true;
+}
+
+bool PolicyManager::removePolicy(const std::string& project,
+                                  const std::string& principal,
+                                  std::string& errorOut) {
+    const std::string key = keyOf(project, principal);
+    {
+        // Removing the last admin of a secured project is the same lockout as
+        // enabling without one, so it is refused for the same reason.
+        std::shared_lock lock(mutex_);
+        auto sec_it = security_.find(project);
+        const bool secured = sec_it != security_.end() && sec_it->second.enabled;
+        auto target = policies_.find(key);
+        if (secured && target != policies_.end() && target->second.admin) {
+            const std::string prefix = project + ":";
+            uint32_t admins = 0;
+            for (const auto& [k, pol] : policies_) {
+                if (k.rfind(prefix, 0) == 0 && pol.admin) ++admins;
+            }
+            if (admins <= 1) {
+                errorOut = "refusing to remove the last admin policy of secured "
+                           "project '" + project + "': it would leave the project "
+                           "unmanageable. Disable security or add another admin "
+                           "first.";
+                return false;
+            }
+        }
+    }
+
+    try {
+        store_.remove(SYSTEM_COLLECTION, key);
+    } catch (const std::exception& e) {
+        errorOut = std::string("failed to remove policy: ") + e.what();
+        return false;
+    }
+    {
+        std::unique_lock lock(mutex_);
+        policies_.erase(key);
+    }
+    return true;
+}
+
+std::optional<Policy> PolicyManager::getPolicy(const std::string& project,
+                                                const std::string& principal) const {
+    std::shared_lock lock(mutex_);
+    auto it = policies_.find(keyOf(project, principal));
+    if (it == policies_.end()) return std::nullopt;
+    return it->second;
+}
+
+std::vector<Policy> PolicyManager::listPolicies(const std::string& project) const {
+    std::shared_lock lock(mutex_);
+    std::vector<Policy> out;
+    const std::string prefix = project + ":";
+    for (const auto& [key, pol] : policies_) {
+        if (key.rfind(prefix, 0) == 0) out.push_back(pol);
+    }
+    return out;
+}
+
+}  // namespace smartbotic::database

service/src/security/policy_manager.hpp

@@ -0,0 +1,186 @@
+// v2.8.0 — per-project row- and column-level access policy.
+//
+// Design and the approved decisions behind it:
+// docs/superpowers/specs/2026-08-08-rls-cls-design.md
+//
+// Shape
+// -----
+// One document per (project, principal) in the global `_policies` system
+// collection, id = "<project>:<principal>". A reserved id
+// "<project>:__security__" carries the per-project enable flag and mode. Using
+// one global collection keyed by qualified id matches how `_views` and
+// `_collection_meta` already work, so nothing new is needed to make policy
+// durable: `_policies` is an ordinary collection and is WAL'd and snapshotted
+// with everything else.
+//
+// Default posture
+// ---------------
+// Security is OFF per project. With it off, `authorize()` allows everything and
+// costs one cached map lookup. Once ON for a project, access inside that project
+// is DENY-BY-DEFAULT: a principal with no matching entry gets nothing. That is
+// what makes an enabled project an actual boundary rather than a suggestion.
+//
+// Two safety properties, both deliberate:
+//
+//   * Lockout is structurally impossible. Enabling security is REFUSED unless
+//     at least one policy in that project has `admin: true`. Care is not a
+//     safety mechanism; a refused enable is.
+//
+//   * `mode: "audit"` evaluates every policy and logs what it WOULD deny, then
+//     allows the request. This is how a live project gets switched on: enable in
+//     audit, watch until the log is quiet, then flip to enforce. Without it,
+//     deny-by-default would strand existing consumers the instant it was armed.
+
+#pragma once
+
+#include <atomic>
+#include <mutex>
+#include <optional>
+#include <shared_mutex>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "document.hpp"
+
+namespace smartbotic::database {
+
+class MemoryStore;
+
+// What a principal may do with one collection or file type.
+struct PolicyRule {
+    bool read = false;
+    bool write = false;
+    // Column mask: dot-notation paths removed from every document returned.
+    // Same path syntax as a view's `exclude`, and applied with the same
+    // applyProjection() helper.
+    std::vector<std::string> mask;
+    // Row predicate, AND-merged with the caller's filters exactly as a view's
+    // `where` already is.
+    std::vector<Filter> row;
+};
+
+struct Policy {
+    std::string principal;
+    // admin grants read+write on everything in the project plus the right to
+    // modify `_policies`. At least one admin is required to enable security.
+    bool admin = false;
+    // Keyed by collection name (bare, within the project) or file type.
+    // "*" is a fallback; an exact match always wins over it.
+    std::unordered_map<std::string, PolicyRule> collections;
+    std::unordered_map<std::string, PolicyRule> files;
+    uint64_t updatedAt = 0;
+};
+
+// Set to true ONLY when the access gate is called from every handler that can
+// read or write user data. Until then `setSecurity(enabled=true)` is refused,
+// because a project enforcing on some paths and silently allowing on others
+// reports protection it does not have - strictly worse than security being off.
+//
+// Gated so far: Get, Count.
+// Still to wire: Find, Exists, SimilaritySearch, GetVersionHistory,
+//   GetDocumentVersion, RestoreVersion, RestoreToDate, GetCollectionInfo,
+//   Insert, Update, PatchDocument, Delete, Subscribe/Watch,
+//   UploadFile, DownloadFile, DeleteFile, GetFileInfo, ListFiles.
+//
+// Also outstanding: masked-field application on returned documents at each
+// read site, refusing writes that touch a masked field, and the management
+// surface (dedicated RPCs + CLI) so policy edits refresh this cache and cannot
+// bypass the lockout guards by writing `_policies` directly.
+inline constexpr bool kEnforcementCoverageComplete = false;
+
+enum class SecurityMode { Enforce, Audit };
+
+struct ProjectSecurity {
+    bool enabled = false;
+    SecurityMode mode = SecurityMode::Enforce;
+};
+
+// The answer handlers act on.
+struct Decision {
+    bool allowed = true;
+    // Populated only when allowed. Empty mask means no columns removed.
+    std::vector<std::string> mask;
+    std::vector<Filter> row;
+    // Set when the decision was a denial that audit mode converted to an
+    // allow. Handlers ignore it; it exists so the caller can be told nothing
+    // while the operator sees everything.
+    bool audited_denial = false;
+    std::string reason;
+};
+
+enum class Access { Read, Write };
+
+class PolicyManager {
+public:
+    static constexpr const char* SYSTEM_COLLECTION = "_policies";
+    static constexpr const char* SECURITY_DOC = "__security__";
+
+    explicit PolicyManager(MemoryStore& store);
+
+    // Load every policy into cache. Call once at startup, after recovery.
+    void loadFromStore();
+
+    // Hot path. Returns allow-everything when the project has security off,
+    // which is the default and the case for every pre-2.8.0 deployment.
+    [[nodiscard]] Decision authorize(const std::string& project,
+                                     const std::string& principal,
+                                     const std::string& collection,
+                                     Access access) const;
+
+    // Same evaluation against a file type rather than a collection.
+    [[nodiscard]] Decision authorizeFile(const std::string& project,
+                                         const std::string& principal,
+                                         const std::string& fileType,
+                                         Access access) const;
+
+    // True when `principal` is an admin of `project`, or when the project has
+    // security disabled (in which case everyone may manage policy).
+    [[nodiscard]] bool mayAdminister(const std::string& project,
+                                     const std::string& principal) const;
+
+    [[nodiscard]] ProjectSecurity securityOf(const std::string& project) const;
+
+    // Refused when enabling without at least one admin policy in the project.
+    bool setSecurity(const std::string& project, const ProjectSecurity& sec,
+                     std::string& errorOut);
+
+    bool setPolicy(const std::string& project, const Policy& policy,
+                   std::string& errorOut);
+    bool removePolicy(const std::string& project, const std::string& principal,
+                      std::string& errorOut);
+    [[nodiscard]] std::optional<Policy> getPolicy(const std::string& project,
+                                                  const std::string& principal) const;
+    [[nodiscard]] std::vector<Policy> listPolicies(const std::string& project) const;
+
+    // Test seam. Lets this component's own tests exercise ENFORCING behaviour
+    // while kEnforcementCoverageComplete is still false. Nothing on the RPC
+    // path calls it, so it cannot be used to arm a half-wired deployment.
+    void allowEnableForTests() { allow_enable_ = true; }
+
+    // True if any project has security enabled. Lets handlers skip the
+    // per-request work entirely on installations that never turned it on.
+    [[nodiscard]] bool anyProjectSecured() const noexcept {
+        return secured_count_.load(std::memory_order_relaxed) > 0;
+    }
+
+private:
+    MemoryStore& store_;
+    mutable std::shared_mutex mutex_;
+    // key: "<project>:<principal>"
+    std::unordered_map<std::string, Policy> policies_;
+    std::unordered_map<std::string, ProjectSecurity> security_;
+    std::atomic<uint32_t> secured_count_{0};
+    bool allow_enable_ = kEnforcementCoverageComplete;
+
+    [[nodiscard]] Decision evaluate(
+        const std::string& project,
+        const std::string& principal,
+        const std::string& target,
+        Access access,
+        bool is_file) const;
+
+    void recountSecured();
+};
+
+}  // namespace smartbotic::database

tests/CMakeLists.txt

@@ -583,3 +583,43 @@ find_package(Threads REQUIRED)
 target_link_libraries(test_file_project_scope PRIVATE Threads::Threads)
 
 add_test(NAME test_file_project_scope COMMAND test_file_project_scope)
+
+# v2.8.0 — access policy engine. The critical assertions are the safety ones:
+# security off allows everything (so upgrades are inert), enabling is refused
+# without an admin policy, the last admin cannot be removed, and audit mode
+# evaluates honestly but allows.
+add_executable(test_policy_manager
+    test_policy_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/security/policy_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/memory_store.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/config/collection_config_manager.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/history_store.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/persistence/wal.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/json_parse.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src/doc_binary.cpp
+)
+
+target_include_directories(test_policy_manager PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/../service/src
+    ${yyjson_INCLUDE_DIRS}
+)
+
+target_link_libraries(test_policy_manager PRIVATE ${yyjson_LIBRARIES})
+
+if(TARGET nlohmann_json::nlohmann_json)
+    target_link_libraries(test_policy_manager PRIVATE nlohmann_json::nlohmann_json)
+else()
+    target_include_directories(test_policy_manager PRIVATE ${NLOHMANN_JSON_INCLUDE_DIRS})
+endif()
+
+if(TARGET spdlog::spdlog)
+    target_link_libraries(test_policy_manager PRIVATE spdlog::spdlog)
+else()
+    target_link_libraries(test_policy_manager PRIVATE ${SPDLOG_LIBRARIES})
+    target_include_directories(test_policy_manager PRIVATE ${SPDLOG_INCLUDE_DIRS})
+endif()
+
+find_package(Threads REQUIRED)
+target_link_libraries(test_policy_manager PRIVATE Threads::Threads)
+
+add_test(NAME test_policy_manager COMMAND test_policy_manager)

tests/test_policy_manager.cpp

@@ -0,0 +1,342 @@
+// v2.8.0 — access policy engine tests.
+//
+// The properties that matter most here are the safety ones, because they are
+// what make the feature usable on a live system rather than a footgun:
+//
+//   * security OFF is the default and allows everything, so upgrading changes
+//     nothing for any existing deployment;
+//   * enabling is REFUSED without an admin policy, which makes lockout
+//     structurally impossible rather than a matter of operator care;
+//   * removing the last admin of a secured project is refused for the same
+//     reason;
+//   * audit mode evaluates honestly and logs, but allows - the migration path
+//     for switching a live project on;
+//   * once enforcing, an unlisted principal gets nothing (deny-by-default), and
+//     system collections are admin-only, since read access to `_policies` would
+//     expose the whole access model and write access would be escalation.
+
+#include <atomic>
+#include <filesystem>
+#include <iostream>
+#include <string>
+#include <unistd.h>
+
+#include <nlohmann/json.hpp>
+
+#include "memory_store.hpp"
+#include "security/policy_manager.hpp"
+
+// setSecurity(enabled=true) is refused while kEnforcementCoverageComplete is
+// false - see policy_manager.hpp. The fixture calls allowEnableForTests() so
+// these tests can still exercise enforcing behaviour; a separate test below
+// asserts the guard itself.
+
+namespace fs = std::filesystem;
+using namespace smartbotic::database;
+
+namespace {
+
+int g_pass = 0;
+int g_fail = 0;
+
+void check(bool cond, const char* msg) {
+    if (cond) {
+        ++g_pass;
+    } else {
+        ++g_fail;
+        std::cerr << "FAIL: " << msg << "\n";
+    }
+}
+
+struct Fixture {
+    MemoryStore store;
+    PolicyManager pm;
+
+    Fixture() : store(MemoryStore::Config{}), pm(store) {
+        store.start();
+        pm.loadFromStore();
+        // These tests are the engine's own; they must be able to reach
+        // enforcing behaviour even though the deployment-level guard is armed.
+        pm.allowEnableForTests();
+    }
+    ~Fixture() { store.stop(); }
+};
+
+Policy mkPolicy(const std::string& principal, bool admin = false) {
+    Policy p;
+    p.principal = principal;
+    p.admin = admin;
+    return p;
+}
+
+PolicyRule rw(bool r, bool w) {
+    PolicyRule x;
+    x.read = r;
+    x.write = w;
+    return x;
+}
+
+// -------------------------------------------------------------------------
+
+void test_security_off_allows_everything() {
+    Fixture f;
+    // No policies, no security record: the pre-2.8.0 world.
+    auto d = f.pm.authorize("acme", "nobody", "users", Access::Read);
+    check(d.allowed, "security off allows an unknown principal to read");
+    check(f.pm.authorize("acme", "nobody", "users", Access::Write).allowed,
+          "security off allows writes too");
+    check(f.pm.authorize("acme", "nobody", "_policies", Access::Read).allowed,
+          "security off does not even guard system collections");
+    check(!f.pm.anyProjectSecured(), "no project is secured by default");
+    check(f.pm.mayAdminister("acme", "anyone"),
+          "with security off anyone may create the first policy");
+}
+
+void test_enabling_without_admin_is_refused() {
+    Fixture f;
+    std::string err;
+    ProjectSecurity sec;
+    sec.enabled = true;
+
+    check(!f.pm.setSecurity("acme", sec, err),
+          "enabling security with no admin policy must be refused");
+    check(err.find("admin") != std::string::npos,
+          "and the error must say why - no admin policy");
+    check(!f.pm.securityOf("acme").enabled, "security stays off after refusal");
+
+    // Non-admin policies are not enough.
+    auto p = mkPolicy("shadowman");
+    p.collections["users"] = rw(true, false);
+    check(f.pm.setPolicy("acme", p, err), "a non-admin policy can be created");
+    check(!f.pm.setSecurity("acme", sec, err),
+          "a non-admin policy does not satisfy the admin requirement");
+
+    check(f.pm.setPolicy("acme", mkPolicy("ops", /*admin=*/true), err),
+          "an admin policy can be created");
+    check(f.pm.setSecurity("acme", sec, err),
+          "with an admin present, enabling succeeds");
+    check(f.pm.securityOf("acme").enabled, "and security is now on");
+    check(f.pm.anyProjectSecured(), "anyProjectSecured reflects it");
+}
+
+void test_deny_by_default_once_enforcing() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto reader = mkPolicy("reader");
+    reader.collections["users"] = rw(true, false);
+    f.pm.setPolicy("acme", reader, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(!f.pm.authorize("acme", "stranger", "users", Access::Read).allowed,
+          "a principal with no policy gets nothing");
+    check(f.pm.authorize("acme", "reader", "users", Access::Read).allowed,
+          "a granted read is allowed");
+    check(!f.pm.authorize("acme", "reader", "users", Access::Write).allowed,
+          "read does not imply write");
+    check(!f.pm.authorize("acme", "reader", "sessions", Access::Read).allowed,
+          "a collection with no rule is denied even for a known principal");
+    check(f.pm.authorize("acme", "ops", "anything", Access::Write).allowed,
+          "admin may write anything in the project");
+
+    // Another project is untouched.
+    check(f.pm.authorize("other", "stranger", "users", Access::Read).allowed,
+          "enabling one project must not affect another");
+}
+
+void test_wildcard_rule_and_exact_match_precedence() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    p.collections["*"] = rw(true, false);
+    p.collections["secrets"] = rw(false, false);
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(f.pm.authorize("acme", "svc", "anything", Access::Read).allowed,
+          "the * fallback grants read on an unlisted collection");
+    check(!f.pm.authorize("acme", "svc", "secrets", Access::Read).allowed,
+          "an exact rule overrides the * fallback, even to deny");
+}
+
+void test_system_collections_are_admin_only() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    p.collections["*"] = rw(true, true);
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(!f.pm.authorize("acme", "svc", "_policies", Access::Read).allowed,
+          "a * grant must NOT reach _policies - that would expose the access model");
+    check(!f.pm.authorize("acme", "svc", "_policies", Access::Write).allowed,
+          "nor allow writing it - that would be privilege escalation");
+    check(!f.pm.authorize("acme", "svc", "_views", Access::Read).allowed,
+          "no system collection is reachable via *");
+    check(f.pm.authorize("acme", "ops", "_policies", Access::Write).allowed,
+          "an admin may still manage system collections");
+    check(!f.pm.mayAdminister("acme", "svc"), "a non-admin may not administer policy");
+    check(f.pm.mayAdminister("acme", "ops"), "an admin may");
+}
+
+void test_mask_and_row_predicate_are_returned() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    PolicyRule r = rw(true, false);
+    r.mask = {"ssn", "profile.dob"};
+    Filter tenant;
+    tenant.field = "tenant";
+    tenant.op = FilterOp::EQ;
+    tenant.value = "acme";
+    r.row = {tenant};
+    p.collections["users"] = r;
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    auto d = f.pm.authorize("acme", "svc", "users", Access::Read);
+    check(d.allowed, "granted");
+    check(d.mask.size() == 2, "the column mask comes back with the decision");
+    check(d.mask[0] == "ssn", "mask paths preserved");
+    check(d.row.size() == 1 && d.row[0].field == "tenant",
+          "the row predicate comes back so the handler can AND-merge it");
+}
+
+void test_audit_mode_logs_but_allows() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec;
+    sec.enabled = true;
+    sec.mode = SecurityMode::Audit;
+    check(f.pm.setSecurity("acme", sec, err), "audit mode can be enabled");
+
+    auto d = f.pm.authorize("acme", "stranger", "users", Access::Read);
+    check(d.allowed, "audit mode ALLOWS what enforce mode would deny");
+    check(d.audited_denial, "but records that it was really a denial");
+    check(!d.reason.empty(), "and keeps the reason for the operator log");
+
+    // Flip to enforce and the same request is refused.
+    sec.mode = SecurityMode::Enforce;
+    f.pm.setSecurity("acme", sec, err);
+    auto d2 = f.pm.authorize("acme", "stranger", "users", Access::Read);
+    check(!d2.allowed, "enforce mode denies the same request");
+    check(!d2.audited_denial, "and does not mark it as merely audited");
+}
+
+void test_cannot_remove_last_admin_of_secured_project() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(!f.pm.removePolicy("acme", "ops", err),
+          "removing the last admin of a secured project must be refused");
+    check(err.find("last admin") != std::string::npos, "with a clear reason");
+
+    check(f.pm.setPolicy("acme", mkPolicy("ops2", true), err), "add a second admin");
+    check(f.pm.removePolicy("acme", "ops", err),
+          "now the first admin can be removed");
+    check(!f.pm.removePolicy("acme", "ops2", err),
+          "but not the remaining last one");
+}
+
+void test_policies_survive_reload() {
+    Fixture f;
+    std::string err;
+    auto p = mkPolicy("svc");
+    PolicyRule r = rw(true, false);
+    r.mask = {"ssn"};
+    p.collections["users"] = r;
+    f.pm.setPolicy("acme", p, err);
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec; sec.enabled = true; sec.mode = SecurityMode::Audit;
+    f.pm.setSecurity("acme", sec, err);
+
+    // Reload from the same store, as a restart would.
+    f.pm.loadFromStore();
+
+    check(f.pm.securityOf("acme").enabled, "enable flag survives reload");
+    check(f.pm.securityOf("acme").mode == SecurityMode::Audit,
+          "mode survives reload");
+    auto got = f.pm.getPolicy("acme", "svc");
+    check(got.has_value(), "policy survives reload");
+    check(got && got->collections.count("users") == 1, "rules survive reload");
+    check(got && got->collections["users"].mask.size() == 1,
+          "the column mask survives reload");
+    check(f.pm.listPolicies("acme").size() == 2, "both policies listed");
+    check(f.pm.listPolicies("other").empty(),
+          "listing is scoped to the project");
+}
+
+void test_file_rules_are_separate_from_collections() {
+    Fixture f;
+    std::string err;
+    f.pm.setPolicy("acme", mkPolicy("ops", true), err);
+    auto p = mkPolicy("svc");
+    p.collections["*"] = rw(true, true);
+    p.files["plugin"] = rw(true, false);
+    f.pm.setPolicy("acme", p, err);
+    ProjectSecurity sec; sec.enabled = true;
+    f.pm.setSecurity("acme", sec, err);
+
+    check(f.pm.authorizeFile("acme", "svc", "plugin", Access::Read).allowed,
+          "a granted file type reads");
+    check(!f.pm.authorizeFile("acme", "svc", "plugin", Access::Write).allowed,
+          "file write is separately gated");
+    check(!f.pm.authorizeFile("acme", "svc", "document", Access::Read).allowed,
+          "an unlisted file type is denied - a collection * does not cover files");
+}
+
+// The deployment guard itself: without the test seam, arming security must be
+// refused while enforcement is not wired into every handler. A project
+// protected on some paths and open on others reports safety it does not have.
+void test_enable_is_refused_while_coverage_incomplete() {
+    MemoryStore store(MemoryStore::Config{});
+    store.start();
+    PolicyManager pm(store);          // note: NO allowEnableForTests()
+    pm.loadFromStore();
+    std::string err;
+    pm.setPolicy("acme", mkPolicy("ops", true), err);
+    ProjectSecurity sec; sec.enabled = true;
+
+    if (kEnforcementCoverageComplete) {
+        check(pm.setSecurity("acme", sec, err),
+              "coverage complete: enabling is permitted");
+    } else {
+        check(!pm.setSecurity("acme", sec, err),
+              "coverage incomplete: enabling must be refused even with an admin");
+        check(err.find("not yet wired") != std::string::npos,
+              "and the refusal must say enforcement is incomplete");
+        check(!pm.securityOf("acme").enabled, "security stays off");
+    }
+    store.stop();
+}
+
+}  // namespace
+
+int main() {
+    std::cout << "=== test_policy_manager ===\n";
+    test_security_off_allows_everything();
+    test_enabling_without_admin_is_refused();
+    test_deny_by_default_once_enforcing();
+    test_wildcard_rule_and_exact_match_precedence();
+    test_system_collections_are_admin_only();
+    test_mask_and_row_predicate_are_returned();
+    test_audit_mode_logs_but_allows();
+    test_cannot_remove_last_admin_of_secured_project();
+    test_policies_survive_reload();
+    test_file_rules_are_separate_from_collections();
+    test_enable_is_refused_while_coverage_incomplete();
+
+    std::cout << "passed: " << g_pass << ", failed: " << g_fail << "\n";
+    return g_fail == 0 ? 0 : 1;
+}