diff --git a/openmirai/mongo/change_stream.go b/openmirai/mongo/change_stream.go
index f86e9e0fd..c2e18ef67 100644
--- a/openmirai/mongo/change_stream.go
+++ b/openmirai/mongo/change_stream.go
@@ -503,7 +503,7 @@ func (stream *ChangeStream) validateEventOwnership(operationType string, fullDoc
 	}
 	if len(fullDocument) > 0 {
 		value, ok := lookupChangePath(fullDocument, stream.policy.TenantPath)
-		if !ok || value.Type != bson.TypeString || value.StringValue() != stream.scope.TenantID {
+		if !ok || !tenantScopeMatches(value, stream.policy, stream.scope.TenantID) {
 			return newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
 		return nil
@@ -512,7 +512,7 @@ func (stream *ChangeStream) validateEventOwnership(operationType string, fullDoc
 		return nil
 	}
 	value, ok := lookupChangePath(documentKey, stream.policy.TenantPath)
-	if !ok || value.Type != bson.TypeString || value.StringValue() != stream.scope.TenantID {
+	if !ok || !tenantScopeMatches(value, stream.policy, stream.scope.TenantID) {
 		return newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 	}
 	return nil
diff --git a/openmirai/mongo/collection.go b/openmirai/mongo/collection.go
index 77ebe3d5f..76b2cec55 100644
--- a/openmirai/mongo/collection.go
+++ b/openmirai/mongo/collection.go
@@ -885,7 +885,7 @@ func prependAggregateScopeMatch(value bson.A, policy PolicyRow, scope OperationS
 		if strings.TrimSpace(policy.TenantPath) == "" || scope.TenantID == "" {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
-		owned = append(owned, bson.E{Key: policy.TenantPath, Value: scope.TenantID})
+		owned = append(owned, bson.E{Key: policy.TenantPath, Value: tenantScopeValue(policy, scope.TenantID)})
 	}
 	owned = append(owned, bson.E{Key: "_pii.state", Value: "active"})
 	result := make(bson.A, 0, len(value)+1)
diff --git a/openmirai/mongo/enroll.go b/openmirai/mongo/enroll.go
index 0990848ac..559552364 100644
--- a/openmirai/mongo/enroll.go
+++ b/openmirai/mongo/enroll.go
@@ -10,6 +10,8 @@ package mongo
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"time"
 
@@ -157,13 +159,14 @@ func EnrollActiveRecords(ctx context.Context, config RuntimeConfig, scope Operat
 				continue
 			}
 			index := field.ExactBlindIndex.Value
+			scope := scopeBytes(scope)
 			bikQuery := keys.ActiveQuery{
 				RecordType: keys.RecordTypeBIK, Environment: string(snapshot.Environment), Service: snapshot.ServiceID,
 				LogicalKeyDatabase: snapshot.KeyDatabase, Region: snapshot.LogicalRegion,
-				LogicalResidency: string(snapshot.Residency), TenantScope: []byte(scope.TenantID),
+				LogicalResidency: string(snapshot.Residency), TenantScope: scope,
 				Database: snapshot.ApplicationDatabase, Collection: collectionName, Purpose: index.Purpose,
 			}
-			recordID := "poc-bik-" + index.IndexID
+			recordID := enrollmentBIKRecordID(collectionName, index.IndexID, scope)
 			bikID, bikSkipped, bikErr := enrollActive(ctx, manager, bikQuery, func() (keys.KeyRecord, error) {
 				input := enrollmentRecordBase(snapshot, recordID, keys.RecordTypeBIK, keys.ChildBIK, index.Purpose)
 				input.ParentRef, input.KEKRef = kekID, kekID
@@ -171,7 +174,7 @@ func EnrollActiveRecords(ctx context.Context, config RuntimeConfig, scope Operat
 				input.ParentGeneration, input.ChildGeneration, input.BIKGeneration = 1, 1, 1
 				input.Database = snapshot.ApplicationDatabase
 				input.Collection = collectionName
-				input.TenantScope = []byte(scope.TenantID)
+				input.TenantScope = scope
 				input.DataClass = string(field.DataClass)
 				input.QueryPurpose = index.Purpose
 				input.NormalizationVersion = index.Version
@@ -210,6 +213,11 @@ func EnrollActiveRecords(ctx context.Context, config RuntimeConfig, scope Operat
 	return report, nil
 }
 
+func enrollmentBIKRecordID(collectionName, indexID string, scope []byte) string {
+	scopeDigest := sha256.Sum256(scope)
+	return "poc-bik-" + sanitizeIndexName(collectionName) + "-" + sanitizeIndexName(indexID) + "-" + hex.EncodeToString(scopeDigest[:])
+}
+
 func sanitizeIndexName(path string) string {
 	result := make([]byte, 0, len(path))
 	for _, char := range path {
diff --git a/openmirai/mongo/enroll_test.go b/openmirai/mongo/enroll_test.go
new file mode 100644
index 000000000..d6c67a027
--- /dev/null
+++ b/openmirai/mongo/enroll_test.go
@@ -0,0 +1,30 @@
+// Copyright 2026 OpenMirai
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+
+package mongo
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestEnrollmentBIKRecordIDIsScopedAndCollectionSpecific(t *testing.T) {
+	first := enrollmentBIKRecordID("users", "EMAIL", []byte("tenant-a"))
+	secondCollection := enrollmentBIKRecordID("learners", "EMAIL", []byte("tenant-a"))
+	secondScope := enrollmentBIKRecordID("users", "EMAIL", []byte("tenant-b"))
+
+	if first == secondCollection || first == secondScope {
+		t.Fatalf("BIK record IDs must distinguish collection and scope: %q", first)
+	}
+	if strings.Contains(first, "tenant-a") {
+		t.Fatalf("BIK record ID must not expose tenant scope: %q", first)
+	}
+	if !strings.HasPrefix(first, "poc-bik-users-EMAIL-") {
+		t.Fatalf("unexpected BIK record ID format: %q", first)
+	}
+}
diff --git a/openmirai/mongo/g020_lane_b_test.go b/openmirai/mongo/g020_lane_b_test.go
index 3fbb13a16..aaed74b19 100644
--- a/openmirai/mongo/g020_lane_b_test.go
+++ b/openmirai/mongo/g020_lane_b_test.go
@@ -15,6 +15,8 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 	drivermongo "go.mongodb.org/mongo-driver/v2/mongo"
 	"go.mongodb.org/mongo-driver/v2/mongo/options"
+	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/compiler"
+	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/compiler/query"
 	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/compiler/update"
 	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/limits"
 )
@@ -56,6 +58,45 @@ func TestG020LaneBAggregatePrependsScopeBeforeEveryCallerPipelineShape(t *testin
 	}
 }
 
+func TestG020LaneBAggregateUnwindCompilerBoundary(t *testing.T) {
+	driver := newG020Driver()
+	client := newG020Client(&g020CompilerPipelineRuntime{}, driver)
+	client.policies["users"] = g020AggregatePolicy()
+	collection, err := client.ApplicationDatabase().Collection("users")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pipeline := bson.A{bson.D{{Key: "$unwind", Value: bson.D{
+		{Key: "path", Value: "$tags"},
+		{Key: "includeArrayIndex", Value: "tagIndex"},
+	}}}}
+	cursor, err := collection.Aggregate(context.Background(), g020Scope(), pipeline, ResultPolicy{ID: "G020_UNWIND"})
+	if err != nil || cursor == nil {
+		t.Fatalf("safe unwind Aggregate = %#v, %v", cursor, err)
+	}
+	if len(driver.aggregatePipeline) != 4 {
+		t.Fatalf("aggregate pipeline = %#v, want scope, input bound, unwind, output bound", driver.aggregatePipeline)
+	}
+	assertG020OwnedAggregateStage(t, driver.aggregatePipeline[0])
+	if stage, ok := driver.aggregatePipeline[1].(bson.D); !ok || len(stage) != 1 || stage[0].Key != "$limit" {
+		t.Fatalf("aggregate input bound = %#v", driver.aggregatePipeline[1])
+	}
+	if stage, ok := driver.aggregatePipeline[2].(bson.D); !ok || len(stage) != 1 || stage[0].Key != "$unwind" {
+		t.Fatalf("aggregate unwind stage = %#v", driver.aggregatePipeline[2])
+	}
+
+	before := len(driver.aggregatePipeline)
+	_, err = collection.Aggregate(context.Background(), g020Scope(), bson.A{bson.D{{Key: "$unwind", Value: "$email"}}}, ResultPolicy{ID: "G020_REJECT"})
+	if !hasG020ErrorCode(err, ErrorUnsupportedOperation) || len(driver.aggregatePipeline) != before {
+		t.Fatalf("protected unwind err=%v pipeline=%#v, want pre-effect unsupported rejection", err, driver.aggregatePipeline)
+	}
+	_, err = collection.Aggregate(context.Background(), g020Scope(), bson.A{bson.D{{Key: "$group", Value: bson.D{{Key: "_id", Value: "$status"}, {Key: "count", Value: bson.D{{Key: "$sum", Value: int32(1)}}}}}}}, ResultPolicy{ID: "G020_GROUP_REJECT"})
+	if !hasG020ErrorCode(err, ErrorUnsupportedOperation) || len(driver.aggregatePipeline) != before {
+		t.Fatalf("group result-shape err=%v pipeline=%#v, want pre-effect unsupported rejection", err, driver.aggregatePipeline)
+	}
+}
+
 func TestG020LaneBWatchUsesEventScopeAndRejectsUnsafeDeleteRequests(t *testing.T) {
 	driver := newG020Driver()
 	client := newG020Client(&g020PipelineRuntime{}, driver)
@@ -337,6 +378,18 @@ func TestG020LaneBPublicAdmissionRejectsEmptyOrWrongPolicyBeforeDriver(t *testin
 	}
 }
 
+type g020CompilerPipelineRuntime struct{ g020PipelineRuntime }
+
+func (*g020CompilerPipelineRuntime) CompilePipeline(_ context.Context, row PolicyRow, scope OperationScope, value any) (bson.A, error) {
+	collection, err := normalizedCollection(row)
+	if err != nil {
+		return nil, err
+	}
+	return compiler.CompileAggregate(value, collection, query.Context{
+		Environment: "TEST", Service: "openmirai", Database: "app", Collection: "users", TenantScope: []byte(scope.TenantID), Limits: limits.DefaultEffective(),
+	})
+}
+
 type g020PipelineRuntime struct{}
 
 type g020BulkPipelineRuntime struct{ g020PipelineRuntime }
@@ -458,6 +511,15 @@ func g020Policy() PolicyRow {
 	return PolicyRow{Version: PolicyVersion1, PolicyID: "USERS", Namespace: "app.users", Scope: PolicyScopeTenant, TenantPath: "tenant.id", DocumentIDPath: "_id", Generation: 1}
 }
 
+func g020AggregatePolicy() PolicyRow {
+	row := g020Policy()
+	row.ProtectedFields = []ProtectedFieldPolicy{{
+		Path: "email", BSONType: ProtectedBSONString, DataClass: DataClassPIIRestricted, EncryptionRequired: true,
+		Algorithm: FieldAlgorithmAES256GCM, Normalization: NormalizationNone, AADProfile: AADProfileOMAAD1FieldV1, MaximumPlaintextBytes: 256,
+	}}
+	return row
+}
+
 func g020Scope() OperationScope {
 	return OperationScope{PolicyID: "USERS", TenantID: "tenant-a"}
 }
diff --git a/openmirai/mongo/internal/compiler/compiler.go b/openmirai/mongo/internal/compiler/compiler.go
index ea24739a3..57bb6c9c7 100644
--- a/openmirai/mongo/internal/compiler/compiler.go
+++ b/openmirai/mongo/internal/compiler/compiler.go
@@ -274,9 +274,12 @@ func CompileDistinct(path string, collection policy.Collection, configured ...li
 	return canonical, nil
 }
 
-// CompileAggregate supports $match, $project, $sort, $limit, and $skip. All
-// other stages are rejected at this private boundary so protected semantics
-// cannot be accidentally reintroduced through a future public method.
+// CompileAggregate supports $match, $project, $sort, $limit, $skip, the
+// bounded safe subset of $unwind, and an envelope-preserving $group form. The
+// group form is deliberately restricted to grouping by the source _id and
+// server-side accumulators over non-protected fields; the compiler injects the
+// authenticated envelope fields into the grouped row so the protected cursor
+// can still verify and decrypt it.
 func CompileAggregate(input any, collection policy.Collection, context query.Context) (result bson.A, err error) {
 	defer func() {
 		if recover() != nil {
@@ -302,6 +305,7 @@ func CompileAggregate(input any, collection policy.Collection, context query.Con
 	}
 	fields := protectedPaths(normalized)
 	result = make(bson.A, 0, len(values)+1)
+	inputBound := false
 	for _, stage := range values {
 		if stage.Type != bson.TypeEmbeddedDocument {
 			return nil, ErrUnsafeAggregate
@@ -342,8 +346,11 @@ func CompileAggregate(input any, collection policy.Collection, context query.Con
 				return nil, ErrUnsafeAggregate
 			}
 			result = append(result, bson.D{{Key: operator, Value: cloneRawValue(value)}})
+			if operator == "$limit" {
+				inputBound = true
+			}
 		case "$set", "$addFields":
-			if touchesUnsafe(value, fields, normalized, "") {
+			if aggregateSetUnsafe(value, fields, normalized) {
 				return nil, ErrUnsafeAggregate
 			}
 			result = append(result, bson.D{{Key: operator, Value: cloneRawValue(value)}})
@@ -352,10 +359,33 @@ func CompileAggregate(input any, collection policy.Collection, context query.Con
 				return nil, ErrUnsafeAggregate
 			}
 			result = append(result, bson.D{{Key: operator, Value: cloneRawValue(value)}})
+		case "$unwind":
+			compiled, compileErr := compileUnwind(value, fields, normalized)
+			if compileErr != nil {
+				return nil, ErrUnsafeAggregate
+			}
+			if !inputBound {
+				result = append(result, bson.D{{Key: "$limit", Value: int64(effective.FindLimit())}})
+			}
+			result = append(result, bson.D{{Key: "$unwind", Value: compiled}})
+			inputBound = false
+		case "$group":
+			compiled, compileErr := compileGroup(value, fields, normalized)
+			if compileErr != nil {
+				return nil, ErrUnsafeAggregate
+			}
+			if !inputBound {
+				result = append(result, bson.D{{Key: "$limit", Value: int64(effective.FindLimit())}})
+			}
+			result = append(result, bson.D{{Key: "$group", Value: compiled}})
+			inputBound = false
 		default:
 			return nil, ErrUnsafeAggregate
 		}
 	}
+	if len(result) > MaxAggregateStages {
+		return nil, ErrCompilerLimit
+	}
 	encoded, err := marshalPipeline(result)
 	if err != nil || uint64(len(encoded)) > effective.BSONDocumentBytes() {
 		return nil, ErrCompilerLimit
@@ -480,6 +510,227 @@ func boundedInteger(operator string, value bson.RawValue, maximum uint32) bool {
 	return false
 }
 
+func compileUnwind(value bson.RawValue, fields map[string]struct{}, collection policy.Collection) (bson.RawValue, error) {
+	switch value.Type {
+	case bson.TypeString:
+		path, err := normalizeUnwindPath(value.StringValue(), true)
+		if err != nil || aggregatePathUnsafe(path, fields, collection) {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		return cloneRawValue(value), nil
+	case bson.TypeEmbeddedDocument:
+		elements, err := bson.Raw(value.Value).Elements()
+		if err != nil || len(elements) == 0 {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		result := make(bson.D, 0, len(elements))
+		seen := make(map[string]struct{}, len(elements))
+		path := ""
+		includeArrayIndex := ""
+		for _, element := range elements {
+			key, keyErr := element.KeyErr()
+			if keyErr != nil {
+				return bson.RawValue{}, ErrUnsafeAggregate
+			}
+			if _, exists := seen[key]; exists {
+				return bson.RawValue{}, ErrUnsafeAggregate
+			}
+			seen[key] = struct{}{}
+			item, valueErr := element.ValueErr()
+			if valueErr != nil {
+				return bson.RawValue{}, ErrUnsafeAggregate
+			}
+			switch key {
+			case "path":
+				if item.Type != bson.TypeString {
+					return bson.RawValue{}, ErrUnsafeAggregate
+				}
+				path, err = normalizeUnwindPath(item.StringValue(), true)
+				if err != nil || aggregatePathUnsafe(path, fields, collection) {
+					return bson.RawValue{}, ErrUnsafeAggregate
+				}
+				result = append(result, bson.E{Key: key, Value: cloneRawValue(item)})
+			case "preserveNullAndEmptyArrays":
+				if item.Type != bson.TypeBoolean {
+					return bson.RawValue{}, ErrUnsafeAggregate
+				}
+				result = append(result, bson.E{Key: key, Value: cloneRawValue(item)})
+			case "includeArrayIndex":
+				if item.Type != bson.TypeString {
+					return bson.RawValue{}, ErrUnsafeAggregate
+				}
+				includeArrayIndex, err = normalizeUnwindPath(item.StringValue(), false)
+				if err != nil || aggregatePathUnsafe(includeArrayIndex, fields, collection) {
+					return bson.RawValue{}, ErrUnsafeAggregate
+				}
+				result = append(result, bson.E{Key: key, Value: cloneRawValue(item)})
+			default:
+				return bson.RawValue{}, ErrUnsafeAggregate
+			}
+		}
+		if path == "" || (includeArrayIndex != "" && policy.PathsOverlap(path, includeArrayIndex)) {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		encoded, err := bson.Marshal(result)
+		if err != nil {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		return bson.RawValue{Type: bson.TypeEmbeddedDocument, Value: encoded}, nil
+	default:
+		return bson.RawValue{}, ErrUnsafeAggregate
+	}
+}
+
+func normalizeUnwindPath(value string, reference bool) (string, error) {
+	if reference {
+		if !strings.HasPrefix(value, "$") || strings.HasPrefix(value, "$$") {
+			return "", ErrUnsafeAggregate
+		}
+		value = strings.TrimPrefix(value, "$")
+	} else if strings.HasPrefix(value, "$") {
+		return "", ErrUnsafeAggregate
+	}
+	if value == "" || value != strings.TrimSpace(value) {
+		return "", ErrUnsafeAggregate
+	}
+	return schema.NormalizePath(value)
+}
+
+func aggregatePathUnsafe(path string, fields map[string]struct{}, collection policy.Collection) bool {
+	return hasReservedSegment(path) || protectedPath(path, fields) ||
+		(collection.TenantPath != "" && policy.PathsOverlap(path, collection.TenantPath)) ||
+		(collection.DocumentIDPath != "" && policy.PathsOverlap(path, collection.DocumentIDPath))
+}
+
+// compileGroup admits only a source-document-preserving group. Grouping by an
+// arbitrary key would synthesize a row that cannot be authenticated against a
+// protected source document, so the only accepted _id expression is $_id.
+func compileGroup(value bson.RawValue, fields map[string]struct{}, collection policy.Collection) (bson.RawValue, error) {
+	if value.Type != bson.TypeEmbeddedDocument {
+		return bson.RawValue{}, ErrUnsafeAggregate
+	}
+	elements, err := bson.Raw(value.Value).Elements()
+	if err != nil || len(elements) == 0 {
+		return bson.RawValue{}, ErrUnsafeAggregate
+	}
+	result := make(bson.D, 0, len(elements)+2)
+	seen := make(map[string]struct{}, len(elements)+2)
+	validID := false
+	for _, element := range elements {
+		key, keyErr := element.KeyErr()
+		if keyErr != nil || key == "" {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		if _, exists := seen[key]; exists || hasReservedSegment(key) && key != "_id" {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		seen[key] = struct{}{}
+		item, valueErr := element.ValueErr()
+		if valueErr != nil {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		if key == "_id" {
+			if item.Type != bson.TypeString || item.StringValue() != "$_id" {
+				return bson.RawValue{}, ErrUnsafeAggregate
+			}
+			validID = true
+			result = append(result, bson.E{Key: key, Value: cloneRawValue(item)})
+			continue
+		}
+		if key == "_pii" || key == "_piiIndex" {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		canonical, pathErr := schema.NormalizePath(key)
+		if pathErr != nil || aggregatePathUnsafe(canonical, fields, collection) {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		if item.Type != bson.TypeEmbeddedDocument {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		accumulators, accumulatorErr := bson.Raw(item.Value).Elements()
+		if accumulatorErr != nil || len(accumulators) != 1 {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		accumulatorKey, keyErr := accumulators[0].KeyErr()
+		accumulatorValue, valueErr := accumulators[0].ValueErr()
+		if keyErr != nil || valueErr != nil || !allowedGroupAccumulator(accumulatorKey) || !aggregateExpressionSafe(accumulatorValue, fields, collection, false) {
+			return bson.RawValue{}, ErrUnsafeAggregate
+		}
+		result = append(result, bson.E{Key: canonical, Value: cloneRawValue(item)})
+	}
+	if !validID {
+		return bson.RawValue{}, ErrUnsafeAggregate
+	}
+	// These fields are required by the authenticated result cursor. They are
+	// taken from the first source row and cannot be supplied by a caller.
+	result = append(result,
+		bson.E{Key: "_pii", Value: bson.D{{Key: "$first", Value: "$_pii"}}},
+		bson.E{Key: "_piiIndex", Value: bson.D{{Key: "$first", Value: "$_piiIndex"}}},
+	)
+	encoded, err := bson.Marshal(result)
+	if err != nil {
+		return bson.RawValue{}, ErrUnsafeAggregate
+	}
+	return bson.RawValue{Type: bson.TypeEmbeddedDocument, Value: encoded}, nil
+}
+
+func allowedGroupAccumulator(operator string) bool {
+	switch operator {
+	case "$first", "$sum", "$push", "$min", "$max":
+		return true
+	default:
+		return false
+	}
+}
+
+func aggregateExpressionSafe(value bson.RawValue, fields map[string]struct{}, collection policy.Collection, internal bool) bool {
+	switch value.Type {
+	case bson.TypeString:
+		text := value.StringValue()
+		if !strings.HasPrefix(text, "$") {
+			return true
+		}
+		if strings.HasPrefix(text, "$$") {
+			return false
+		}
+		path, err := schema.NormalizePath(strings.TrimPrefix(text, "$"))
+		if err != nil {
+			return false
+		}
+		if internal && (path == "_pii" || path == "_piiIndex") {
+			return true
+		}
+		return !aggregatePathUnsafe(path, fields, collection)
+	case bson.TypeEmbeddedDocument:
+		elements, err := bson.Raw(value.Value).Elements()
+		if err != nil || len(elements) != 1 {
+			return false
+		}
+		operator, err := elements[0].KeyErr()
+		if err != nil || (operator != "$multiply" && operator != "$add" && operator != "$subtract" && operator != "$ifNull") {
+			return false
+		}
+		item, err := elements[0].ValueErr()
+		if err != nil {
+			return false
+		}
+		return aggregateExpressionSafe(item, fields, collection, internal)
+	case bson.TypeArray:
+		values, err := bson.Raw(value.Value).Values()
+		if err != nil || len(values) > MaxAggregateStages*MaxAggregateStages {
+			return false
+		}
+		for _, item := range values {
+			if !aggregateExpressionSafe(item, fields, collection, internal) {
+				return false
+			}
+		}
+		return true
+	default:
+		return true
+	}
+}
+
 func selectedLimits(values ...limits.Effective) limits.Effective {
 	for _, value := range values {
 		if value.Valid() {
@@ -489,6 +740,32 @@ func selectedLimits(values ...limits.Effective) limits.Effective {
 	return limits.DefaultEffective()
 }
 
+// aggregateSetUnsafe admits computed fields only when both the destination
+// and every referenced source path are non-protected, non-tenant, and
+// non-reserved. This keeps expression evaluation server-side without giving a
+// caller a way to copy protected plaintext into a new field.
+func aggregateSetUnsafe(value bson.RawValue, fields map[string]struct{}, collection policy.Collection) bool {
+	if value.Type != bson.TypeEmbeddedDocument {
+		return true
+	}
+	elements, err := bson.Raw(value.Value).Elements()
+	if err != nil || len(elements) == 0 {
+		return true
+	}
+	for _, element := range elements {
+		key, keyErr := element.KeyErr()
+		item, valueErr := element.ValueErr()
+		if keyErr != nil || valueErr != nil || strings.HasPrefix(key, "$") {
+			return true
+		}
+		canonical, pathErr := schema.NormalizePath(key)
+		if pathErr != nil || aggregatePathUnsafe(canonical, fields, collection) || !aggregateExpressionSafe(item, fields, collection, false) {
+			return true
+		}
+	}
+	return false
+}
+
 func touchesUnsafe(value bson.RawValue, fields map[string]struct{}, collection policy.Collection, prefix string) bool {
 	return touchesUnsafeAt(value, fields, collection, prefix, 0)
 }
diff --git a/openmirai/mongo/internal/compiler/compiler_test.go b/openmirai/mongo/internal/compiler/compiler_test.go
index 1598a8f64..ec0ffabb8 100644
--- a/openmirai/mongo/internal/compiler/compiler_test.go
+++ b/openmirai/mongo/internal/compiler/compiler_test.go
@@ -92,6 +92,113 @@ func TestAggregateCompilesMatchAndRejectsUnsafeStage(t *testing.T) {
 	}
 }
 
+func TestAggregateCompilesSafeUnwindWithBoundedInput(t *testing.T) {
+	pipeline, err := CompileAggregate(bson.A{bson.D{{Key: "$unwind", Value: bson.D{
+		{Key: "path", Value: "$tags"},
+		{Key: "includeArrayIndex", Value: "tagIndex"},
+		{Key: "preserveNullAndEmptyArrays", Value: true},
+	}}}}, compilerCollection(), compilerQueryContext())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pipeline) != 2 {
+		t.Fatalf("compiled unwind = %#v, want owned input limit and unwind", pipeline)
+	}
+	limit, ok := pipeline[0].(bson.D)
+	var limitValue int64
+	var limitValueOK bool
+	if ok && len(limit) == 1 {
+		limitValue, limitValueOK = limit[0].Value.(int64)
+	}
+	if !ok || len(limit) != 1 || limit[0].Key != "$limit" || !limitValueOK || limitValue == int64(0) {
+		t.Fatalf("owned unwind input bound = %#v", pipeline[0])
+	}
+	unwind, ok := pipeline[1].(bson.D)
+	if !ok || len(unwind) != 1 || unwind[0].Key != "$unwind" {
+		t.Fatalf("compiled unwind stage = %#v", pipeline[1])
+	}
+	unwindValue, ok := unwind[0].Value.(bson.RawValue)
+	if !ok || unwindValue.Type != bson.TypeEmbeddedDocument {
+		t.Fatalf("compiled unwind value = %#v", unwind[0].Value)
+	}
+	compiled := bson.Raw(unwindValue.Value)
+	if compiled.Lookup("path").StringValue() != "$tags" || compiled.Lookup("includeArrayIndex").StringValue() != "tagIndex" || !compiled.Lookup("preserveNullAndEmptyArrays").Boolean() {
+		t.Fatalf("compiled unwind options = %x", compiled)
+	}
+	stringPipeline, err := CompileAggregate(bson.A{bson.D{{Key: "$limit", Value: int32(2)}}, bson.D{{Key: "$unwind", Value: "$tags"}}}, compilerCollection(), compilerQueryContext())
+	if err != nil || len(stringPipeline) != 2 {
+		t.Fatalf("string unwind = %#v, %v", stringPipeline, err)
+	}
+	stringStage := stringPipeline[1].(bson.D)
+	if value, ok := stringStage[0].Value.(bson.RawValue); !ok || value.Type != bson.TypeString || bson.RawValue(value).StringValue() != "$tags" {
+		t.Fatalf("string unwind stage = %#v", stringStage)
+	}
+}
+
+func TestAggregateCompilesEnvelopePreservingGroup(t *testing.T) {
+	pipeline, err := CompileAggregate(bson.A{bson.D{{Key: "$group", Value: bson.D{
+		{Key: "_id", Value: "$_id"},
+		{Key: "total", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$multiply", Value: bson.A{"$statusCount", int32(1)}}}}}},
+	}}}}, compilerCollection(), compilerQueryContext())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pipeline) != 2 {
+		t.Fatalf("compiled group = %#v, want input bound and group", pipeline)
+	}
+	group, ok := pipeline[1].(bson.D)
+	if !ok || len(group) != 1 || group[0].Key != "$group" {
+		t.Fatalf("compiled group stage = %#v", pipeline[1])
+	}
+	value, ok := group[0].Value.(bson.RawValue)
+	if !ok {
+		t.Fatalf("compiled group value = %#v", group[0].Value)
+	}
+	compiled := bson.Raw(value.Value)
+	if compiled.Lookup("_id").StringValue() != "$_id" || compiled.Lookup("_pii").Type != bson.TypeEmbeddedDocument || compiled.Lookup("_piiIndex").Type != bson.TypeEmbeddedDocument {
+		t.Fatalf("compiled group envelope = %x", compiled)
+	}
+}
+
+func TestAggregateCompilesSafeComputedSetAndRejectsProtectedReference(t *testing.T) {
+	pipeline, err := CompileAggregate(bson.A{bson.D{{Key: "$set", Value: bson.D{
+		{Key: "grandTotal", Value: bson.D{{Key: "$add", Value: bson.A{"$subtotal", "$tax"}}}},
+	}}}}, compilerCollection(), compilerQueryContext())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pipeline) != 1 {
+		t.Fatalf("computed set = %#v", pipeline)
+	}
+	if _, err := CompileAggregate(bson.A{bson.D{{Key: "$set", Value: bson.D{
+		{Key: "copy", Value: "$email"},
+	}}}}, compilerCollection(), compilerQueryContext()); !errors.Is(err, ErrUnsafeAggregate) {
+		t.Fatalf("protected computed set error = %v, want %v", err, ErrUnsafeAggregate)
+	}
+}
+
+func TestAggregateRejectsUnsafeUnwindReferencesAndGroupedResults(t *testing.T) {
+	tests := []struct {
+		name  string
+		stage bson.D
+	}{
+		{name: "protected path", stage: bson.D{{Key: "$unwind", Value: "$email"}}},
+		{name: "nested protected path", stage: bson.D{{Key: "$unwind", Value: "$email.value"}}},
+		{name: "tenant path", stage: bson.D{{Key: "$unwind", Value: "$tenant.id"}}},
+		{name: "document id", stage: bson.D{{Key: "$unwind", Value: "$_id"}}},
+		{name: "reserved path", stage: bson.D{{Key: "$unwind", Value: "$_pii.state"}}},
+		{name: "protected array index", stage: bson.D{{Key: "$unwind", Value: bson.D{{Key: "path", Value: "$tags"}, {Key: "includeArrayIndex", Value: "email"}}}}},
+		{name: "group result shape", stage: bson.D{{Key: "$group", Value: bson.D{{Key: "_id", Value: "$status"}, {Key: "count", Value: bson.D{{Key: "$sum", Value: int32(1)}}}}}}},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if _, err := CompileAggregate(bson.A{test.stage}, compilerCollection(), compilerQueryContext()); !errors.Is(err, ErrUnsafeAggregate) {
+				t.Fatalf("unsafe aggregate error = %v, want %v", err, ErrUnsafeAggregate)
+			}
+		})
+	}
+}
+
 func TestAggregateAcceptsEmptyPipeline(t *testing.T) {
 	pipeline, err := CompileAggregate(bson.A{}, compilerCollection(), compilerQueryContext())
 	if err != nil {
diff --git a/openmirai/mongo/internal/policy/policy.go b/openmirai/mongo/internal/policy/policy.go
index 9538910e5..a486e8028 100644
--- a/openmirai/mongo/internal/policy/policy.go
+++ b/openmirai/mongo/internal/policy/policy.go
@@ -165,7 +165,12 @@ func normalizeCollection(input Collection) (Collection, error) {
 	if _, err := schema.ValidatePaths(paths); err != nil {
 		return Collection{}, err
 	}
-	if pathOverlaps(documentIDPath, tenantPath) {
+	// A tenant may be represented by the document itself (for example, an
+	// organization document whose tenant identity is its `_id`). Exact
+	// document/tenant equality is safe because protected fields remain barred
+	// from both paths below. Partial overlap is still rejected so a tenant
+	// container cannot weaken document identity isolation.
+	if documentIDPath != tenantPath && pathOverlaps(documentIDPath, tenantPath) {
 		return Collection{}, ErrImmutablePath
 	}
 	for _, path := range paths {
diff --git a/openmirai/mongo/internal/policy/policy_test.go b/openmirai/mongo/internal/policy/policy_test.go
index fb6bf51e3..90cee1659 100644
--- a/openmirai/mongo/internal/policy/policy_test.go
+++ b/openmirai/mongo/internal/policy/policy_test.go
@@ -129,7 +129,6 @@ func TestNormalizeRejectsImmutableAndNestedReservedPolicyPaths(t *testing.T) {
 	}{
 		{name: "tenant overlaps field", edit: func(value *Collection) { value.Fields[0].Path = "tenant" }},
 		{name: "document id overlaps field", edit: func(value *Collection) { value.Fields[0].Path = "_id.value" }},
-		{name: "tenant overlaps document id", edit: func(value *Collection) { value.TenantPath = "_id" }},
 		{name: "nested reserved field", edit: func(value *Collection) { value.Fields[0].Path = "profile._pii.email" }},
 		{name: "nested reserved tenant", edit: func(value *Collection) { value.TenantPath = "profile._pii.tenant" }},
 	}
@@ -145,6 +144,18 @@ func TestNormalizeRejectsImmutableAndNestedReservedPolicyPaths(t *testing.T) {
 	}
 }
 
+func TestNormalizeAllowsDocumentSelfScopedTenant(t *testing.T) {
+	value := validCollection()
+	value.TenantPath = value.DocumentIDPath
+	registry, err := Normalize(Registry{Version: Version, Rows: []Collection{value}})
+	if err != nil {
+		t.Fatalf("Normalize() error = %v, want self-scoped tenant policy to be valid", err)
+	}
+	if got := registry.Rows[0].TenantPath; got != "_id" {
+		t.Fatalf("tenant path = %q, want _id", got)
+	}
+}
+
 func validCollection() Collection {
 	return Collection{
 		Version:        Version,
diff --git a/openmirai/mongo/operation_coordinator.go b/openmirai/mongo/operation_coordinator.go
index 925ad0669..59a302e19 100644
--- a/openmirai/mongo/operation_coordinator.go
+++ b/openmirai/mongo/operation_coordinator.go
@@ -438,20 +438,24 @@ func scopedOperationFilter(compiled bson.Raw, policy PolicyRow, scope OperationS
 	if compiled.Validate() != nil {
 		return nil, newError(ErrorInvalidArgument, ErrorClassValidation, nil)
 	}
-	elements, err := compiled.Elements()
-	if err != nil {
-		return nil, newError(ErrorInvalidArgument, ErrorClassValidation, nil)
-	}
-	base := make(bson.D, 0, len(elements))
-	for _, element := range elements {
-		base = append(base, bson.E{Key: element.Key(), Value: element.Value()})
+	var base bson.D
+	if err := bson.Unmarshal(compiled, &base); err != nil {
+		return nil, newError(ErrorInvalidArgument, ErrorClassValidation, err)
 	}
-	predicates := bson.A{base}
+	tenantPredicateRewritten := false
 	if policy.Scope == PolicyScopeTenant {
 		if scope.TenantID == "" || strings.TrimSpace(policy.TenantPath) == "" {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
-		predicates = append(predicates, bson.D{{Key: policy.TenantPath, Value: scope.TenantID}})
+		var err error
+		base, tenantPredicateRewritten, err = rewriteScopedTenantPredicate(base, policy, scope.TenantID)
+		if err != nil {
+			return nil, err
+		}
+	}
+	predicates := bson.A{base}
+	if policy.Scope == PolicyScopeTenant && !tenantPredicateRewritten {
+		predicates = append(predicates, bson.D{{Key: policy.TenantPath, Value: tenantScopeValue(policy, scope.TenantID)}})
 	}
 	if active {
 		predicates = append(predicates, bson.D{{Key: "_pii.state", Value: "active"}})
@@ -462,6 +466,63 @@ func scopedOperationFilter(compiled bson.Raw, policy PolicyRow, scope OperationS
 	return bson.Marshal(bson.D{{Key: "$and", Value: predicates}})
 }
 
+// rewriteScopedTenantPredicate normalizes the tenant predicate injected by
+// the query compiler. The compiler operates on string tenant IDs, while the
+// facade must use the policy-owned BSON representation (for example a native
+// ObjectID when the tenant path is _id). A matching native predicate is
+// retained; a mismatching caller predicate fails closed.
+func rewriteScopedTenantPredicate(base bson.D, policy PolicyRow, tenantID string) (bson.D, bool, error) {
+	var rewrite func(any) (bool, error)
+	rewrite = func(value any) (bool, error) {
+		switch typed := value.(type) {
+		case bson.D:
+			matched := false
+			for index := range typed {
+				if typed[index].Key == policy.TenantPath {
+					compilerOwned := false
+					if value, ok := typed[index].Value.(string); ok {
+						compilerOwned = value == tenantID
+					} else {
+						valueType, valueBytes, err := bson.MarshalValue(typed[index].Value)
+						if err != nil {
+							return false, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, err)
+						}
+						value := bson.RawValue{Type: valueType, Value: valueBytes}
+						compilerOwned = tenantScopeMatches(value, policy, tenantID)
+					}
+					if !compilerOwned {
+						return false, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
+					}
+					typed[index].Value = tenantScopeValue(policy, tenantID)
+					matched = true
+					continue
+				}
+				nested, err := rewrite(typed[index].Value)
+				if err != nil {
+					return false, err
+				}
+				matched = matched || nested
+			}
+			return matched, nil
+		case bson.A:
+			matched := false
+			for index := range typed {
+				nested, err := rewrite(typed[index])
+				if err != nil {
+					return false, err
+				}
+				matched = matched || nested
+			}
+			return matched, nil
+		default:
+			return false, nil
+		}
+	}
+
+	matched, err := rewrite(base)
+	return base, matched, err
+}
+
 func stripActivePredicates(raw bson.Raw) (bson.Raw, error) {
 	var strip func(bson.Raw) (bson.D, error)
 	strip = func(input bson.Raw) (bson.D, error) {
@@ -832,7 +893,7 @@ func validateAuthenticatedReplacement(policy PolicyRow, scope OperationScope, ex
 		if !existingOK || !replacementOK || !equalBSONValues(existingValue, replacementValue) {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
-		if policy.Scope == PolicyScopeTenant && (scope.TenantID == "" || existingValue.Type != bson.TypeString || existingValue.StringValue() != scope.TenantID || replacementValue.Type != bson.TypeString || replacementValue.StringValue() != scope.TenantID) {
+		if policy.Scope == PolicyScopeTenant && (scope.TenantID == "" || !tenantScopeMatches(existingValue, policy, scope.TenantID) || !tenantScopeMatches(replacementValue, policy, scope.TenantID)) {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
 	}
@@ -1040,7 +1101,7 @@ func filterByDocumentID(id bson.RawValue, policy PolicyRow, scope OperationScope
 		if scope.TenantID == "" || strings.TrimSpace(policy.TenantPath) == "" {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
-		document = append(document, bson.E{Key: policy.TenantPath, Value: scope.TenantID})
+		document = append(document, bson.E{Key: policy.TenantPath, Value: tenantScopeValue(policy, scope.TenantID)})
 	}
 	return bson.Marshal(document)
 }
@@ -1460,7 +1521,7 @@ func deleteCASFilter(encrypted bson.Raw, row PolicyRow, scope OperationScope) (b
 		if scope.TenantID == "" || strings.TrimSpace(row.TenantPath) == "" {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
-		filter = append(filter, bson.E{Key: row.TenantPath, Value: scope.TenantID})
+		filter = append(filter, bson.E{Key: row.TenantPath, Value: tenantScopeValue(row, scope.TenantID)})
 	}
 	return filter, nil
 }
diff --git a/openmirai/mongo/scope_value.go b/openmirai/mongo/scope_value.go
new file mode 100644
index 000000000..9ec534106
--- /dev/null
+++ b/openmirai/mongo/scope_value.go
@@ -0,0 +1,32 @@
+package mongo
+
+import (
+	"strings"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+// tenantScopeValue is the policy-owned representation rule for tenant
+// predicates. Tenant IDs are strings by default. A policy whose tenant path is
+// also its document ID may use a canonical 24-hex ID as a native ObjectID.
+func tenantScopeValue(policy PolicyRow, tenantID string) any {
+	if strings.TrimSpace(policy.TenantPath) == strings.TrimSpace(policy.DocumentIDPath) {
+		if objectID, err := bson.ObjectIDFromHex(strings.TrimSpace(tenantID)); err == nil {
+			return objectID
+		}
+	}
+	return tenantID
+}
+
+func tenantScopeRawValue(policy PolicyRow, tenantID string) bson.RawValue {
+	typeCode, encoded, err := bson.MarshalValue(tenantScopeValue(policy, tenantID))
+	if err != nil {
+		return bson.RawValue{}
+	}
+	return bson.RawValue{Type: typeCode, Value: encoded}
+}
+
+func tenantScopeMatches(value bson.RawValue, policy PolicyRow, tenantID string) bool {
+	expected := tenantScopeRawValue(policy, tenantID)
+	return expected.Type != bson.Type(0) && equalBSONValues(value, expected)
+}
diff --git a/openmirai/mongo/scope_value_test.go b/openmirai/mongo/scope_value_test.go
new file mode 100644
index 000000000..dfde1cd52
--- /dev/null
+++ b/openmirai/mongo/scope_value_test.go
@@ -0,0 +1,81 @@
+package mongo
+
+import (
+	"testing"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+func TestTenantScopeValueUsesObjectIDForDocumentIDTenant(t *testing.T) {
+	policy := PolicyRow{TenantPath: "_id", DocumentIDPath: "_id"}
+	value := tenantScopeRawValue(policy, "507f1f77bcf86cd799439011")
+	if value.Type != bson.TypeObjectID {
+		t.Fatalf("tenant value type = %v, want ObjectID", value.Type)
+	}
+	if !tenantScopeMatches(value, policy, "507f1f77bcf86cd799439011") {
+		t.Fatal("native ObjectID tenant value did not match")
+	}
+}
+
+func TestTenantScopeValueKeepsStringForOrdinaryTenantPath(t *testing.T) {
+	policy := PolicyRow{TenantPath: "organizationId", DocumentIDPath: "_id"}
+	value := tenantScopeRawValue(policy, "507f1f77bcf86cd799439011")
+	if value.Type != bson.TypeString {
+		t.Fatalf("tenant value type = %v, want string", value.Type)
+	}
+	if !tenantScopeMatches(value, policy, "507f1f77bcf86cd799439011") {
+		t.Fatal("string tenant value did not match")
+	}
+}
+
+func TestScopedOperationFilterCanonicalizesCompilerTenantPredicate(t *testing.T) {
+	const tenantID = "507f1f77bcf86cd799439011"
+	policy := PolicyRow{Scope: PolicyScopeTenant, TenantPath: "_id", DocumentIDPath: "_id"}
+	compiled, err := bson.Marshal(bson.D{{Key: "$and", Value: bson.A{
+		bson.D{},
+		bson.D{{Key: "_id", Value: tenantID}},
+	}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	filtered, err := scopedOperationFilter(compiled, policy, OperationScope{PolicyID: "POLICY", TenantID: tenantID}, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var decoded bson.D
+	if err := bson.Unmarshal(filtered, &decoded); err != nil {
+		t.Fatal(err)
+	}
+	ids, strings := countTenantPredicates(decoded, policy.TenantPath)
+	if ids != 1 || strings != 0 {
+		t.Fatalf("tenant predicates = objectIDs:%d strings:%d document:%v", ids, strings, decoded)
+	}
+}
+
+func countTenantPredicates(value any, path string) (ids, strings int) {
+	switch typed := value.(type) {
+	case bson.D:
+		for _, element := range typed {
+			if element.Key == path {
+				switch element.Value.(type) {
+				case bson.ObjectID:
+					ids++
+				case string:
+					strings++
+				}
+			}
+			childIDs, childStrings := countTenantPredicates(element.Value, path)
+			ids += childIDs
+			strings += childStrings
+		}
+	case bson.A:
+		for _, element := range typed {
+			childIDs, childStrings := countTenantPredicates(element, path)
+			ids += childIDs
+			strings += childStrings
+		}
+	}
+	return ids, strings
+}
diff --git a/openmirai/mongo/typed.go b/openmirai/mongo/typed.go
index e16e25d0e..0bf0e27ec 100644
--- a/openmirai/mongo/typed.go
+++ b/openmirai/mongo/typed.go
@@ -10,7 +10,10 @@ package mongo
 
 import (
 	"context"
+	"errors"
+	"strings"
 
+	"go.mongodb.org/mongo-driver/v2/bson"
 	drivermongo "go.mongodb.org/mongo-driver/v2/mongo"
 	"go.mongodb.org/mongo-driver/v2/mongo/options"
 )
@@ -256,6 +259,71 @@ func (c *PlainCollection[T]) CountDocuments(ctx context.Context, filter any, opt
 	return c.collection.CountDocuments(ctx, filter, opts...)
 }
 
+// EnsureUniqueAscendingIndex creates the named unique ascending index on one
+// plain field. It is intentionally a narrow readiness surface: migration
+// tooling can establish a directory invariant without receiving a raw
+// upstream collection handle or a generic index-management adapter.
+func (c *PlainCollection[T]) EnsureUniqueAscendingIndex(ctx context.Context, fieldName, indexName string) error {
+	if c == nil || c.collection == nil || strings.TrimSpace(fieldName) == "" || strings.TrimSpace(indexName) == "" {
+		return errors.New("plain collection and index identity are required")
+	}
+	_, err := c.collection.Indexes().CreateOne(ctx, drivermongo.IndexModel{
+		Keys:    bson.D{{Key: fieldName, Value: 1}},
+		Options: options.Index().SetName(indexName).SetUnique(true),
+	})
+	return err
+}
+
+// VerifyUniqueAscendingIndex proves that the named unique ascending index is
+// present on one plain field. The v2 cursor remains private to the facade;
+// callers receive only the readiness error.
+func (c *PlainCollection[T]) VerifyUniqueAscendingIndex(ctx context.Context, fieldName, indexName string) error {
+	if c == nil || c.collection == nil || strings.TrimSpace(fieldName) == "" || strings.TrimSpace(indexName) == "" {
+		return errors.New("plain collection and index identity are required")
+	}
+	cursor, err := c.collection.Indexes().List(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = cursor.Close(ctx) }()
+	for cursor.Next(ctx) {
+		var index struct {
+			Name   string `bson:"name"`
+			Key    bson.D `bson:"key"`
+			Unique bool   `bson:"unique"`
+		}
+		if err := cursor.Decode(&index); err != nil {
+			return err
+		}
+		if index.Name != indexName {
+			continue
+		}
+		if !index.Unique || len(index.Key) != 1 || index.Key[0].Key != fieldName || !isAscendingIndexValue(index.Key[0].Value) {
+			return errors.New("plain collection unique ascending index is invalid")
+		}
+		return nil
+	}
+	if err := cursor.Err(); err != nil {
+		return err
+	}
+	return errors.New("plain collection unique ascending index is missing")
+}
+
+func isAscendingIndexValue(value any) bool {
+	switch typed := value.(type) {
+	case int:
+		return typed == 1
+	case int32:
+		return typed == 1
+	case int64:
+		return typed == 1
+	case float64:
+		return typed == 1
+	default:
+		return false
+	}
+}
+
 // EstimatedDocumentCount estimates the plain document count.
 func (c *PlainCollection[T]) EstimatedDocumentCount(ctx context.Context, opts ...options.Lister[options.EstimatedDocumentCountOptions]) (int64, error) {
 	return c.collection.EstimatedDocumentCount(ctx, opts...)
diff --git a/openmirai/mongo/typed_plain_index_test.go b/openmirai/mongo/typed_plain_index_test.go
new file mode 100644
index 000000000..4d408506e
--- /dev/null
+++ b/openmirai/mongo/typed_plain_index_test.go
@@ -0,0 +1,23 @@
+package mongo
+
+import (
+	"context"
+	"testing"
+)
+
+type typedPlainIndexTestModel struct {
+	PlainMarker
+}
+
+func TestPlainCollectionIndexReadinessFailsClosedWithoutHandle(t *testing.T) {
+	var collection *PlainCollection[typedPlainIndexTestModel]
+	if err := collection.EnsureUniqueAscendingIndex(context.Background(), "token", "token_unique"); err == nil {
+		t.Fatal("EnsureUniqueAscendingIndex accepted a nil plain handle")
+	}
+	if err := collection.VerifyUniqueAscendingIndex(context.Background(), "token", "token_unique"); err == nil {
+		t.Fatal("VerifyUniqueAscendingIndex accepted a nil plain handle")
+	}
+	if err := collection.VerifyUniqueAscendingIndex(context.Background(), "", "token_unique"); err == nil {
+		t.Fatal("VerifyUniqueAscendingIndex accepted an empty field name")
+	}
+}