diff --git a/providers/aws/keyspaces/keyspaces.go b/providers/aws/keyspaces/keyspaces.go
index 9d4adf587..bfff6e452 100644
--- a/providers/aws/keyspaces/keyspaces.go
+++ b/providers/aws/keyspaces/keyspaces.go
@@ -7,6 +7,7 @@ package keyspaces
 import (
 	"context"
 	"fmt"
+	"regexp"
 	"strings"
 	"sync"
 
@@ -57,7 +58,7 @@ func New(opts *config.Options) *Mock {
 		opts:        opts,
 	}
 
-	for _, sys := range []string{"system", "system_schema", "system_multiregion_info"} {
+	for _, sys := range systemKeyspaceNames() {
 		m.keyspaces.Set(sys, ksdriver.Keyspace{
 			Name: sys, ARN: m.keyspaceARN(sys), ReplicationStrategy: ksdriver.SingleRegion,
 			ReplicationRegions: []string{opts.Region},
@@ -79,13 +80,51 @@ func tableKey(keyspace, table string) string { return keyspace + "/" + table }
 
 func typeKey(keyspace, name string) string { return keyspace + "/" + name }
 
+// maxNameLen is the documented maximum length of keyspace, table, and
+// user-defined type names (API reference KeyspaceName/TableName/TypeName:
+// "Maximum length of 48"; devguide quotas: "Max UDT name length 48").
+const maxNameLen = 48
+
+// keyspaceTableNamePattern is the documented KeyspaceName/TableName pattern
+// ([a-zA-Z0-9][a-zA-Z0-9_]{0,47}), anchored: alphanumerics and underscores
+// only, not starting with an underscore. A leading digit is allowed.
+var keyspaceTableNamePattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_]{0,47}$`)
+
+// validName validates a keyspace or table name against the documented
+// KeyspaceName/TableName constraints.
 func validName(kind, name string) error {
 	if name == "" {
 		return cerrors.Newf(cerrors.InvalidArgument, "%s name is required", kind)
 	}
 
+	if len(name) > maxNameLen {
+		return cerrors.Newf(cerrors.InvalidArgument,
+			"%s name %q must be at most %d characters", kind, name, maxNameLen)
+	}
+
+	if !keyspaceTableNamePattern.MatchString(name) {
+		return cerrors.Newf(cerrors.InvalidArgument,
+			"%s name %q must match pattern [a-zA-Z0-9][a-zA-Z0-9_]{0,47}", kind, name)
+	}
+
+	return nil
+}
+
+// validTypeName validates a user-defined type name. The API model gives
+// TypeName only a 1-48 length constraint (no pattern), so only emptiness,
+// length, and the '/' used in internal store keys are checked.
+func validTypeName(name string) error {
+	if name == "" {
+		return cerrors.New(cerrors.InvalidArgument, "type name is required")
+	}
+
+	if len(name) > maxNameLen {
+		return cerrors.Newf(cerrors.InvalidArgument,
+			"type name %q must be at most %d characters", name, maxNameLen)
+	}
+
 	if strings.Contains(name, "/") {
-		return cerrors.Newf(cerrors.InvalidArgument, "%s name %q must not contain '/'", kind, name)
+		return cerrors.Newf(cerrors.InvalidArgument, "type name %q must not contain '/'", name)
 	}
 
 	return nil
@@ -120,6 +159,10 @@ func (m *Mock) CreateKeyspace(_ context.Context, cfg ksdriver.CreateKeyspaceConf
 		return nil, err
 	}
 
+	if err := validateTags(cfg.Tags); err != nil {
+		return nil, err
+	}
+
 	strategy := cfg.ReplicationStrategy
 	if strategy == "" {
 		strategy = ksdriver.SingleRegion
@@ -244,13 +287,15 @@ func (m *Mock) DeleteKeyspace(_ context.Context, name string) error {
 	return nil
 }
 
+// systemKeyspaceNames lists the four read-only system keyspaces every account
+// has ("Amazon Keyspaces uses four system keyspaces", devguide
+// working-with-keyspaces.html).
+func systemKeyspaceNames() []string {
+	return []string{"system", "system_schema", "system_schema_mcs", "system_multiregion_info"}
+}
+
 func isSystemKeyspace(name string) bool {
-	switch name {
-	case "system", "system_schema", "system_multiregion_info":
-		return true
-	default:
-		return false
-	}
+	return containsStr(systemKeyspaceNames(), name)
 }
 
 func (m *Mock) keyspaceHasTables(keyspace string) bool {
diff --git a/providers/aws/keyspaces/keyspaces_test.go b/providers/aws/keyspaces/keyspaces_test.go
index 402d6c0e9..aa1135c89 100644
--- a/providers/aws/keyspaces/keyspaces_test.go
+++ b/providers/aws/keyspaces/keyspaces_test.go
@@ -69,8 +69,8 @@ func TestKeyspaceLifecycle(t *testing.T) {
 		t.Fatalf("ListKeyspaces: %v", err)
 	}
 
-	if len(list) != 4 {
-		t.Fatalf("expected 4 keyspaces, got %d", len(list))
+	if len(list) != 5 { // 4 system + app
+		t.Fatalf("expected 5 keyspaces, got %d", len(list))
 	}
 }
 
@@ -114,11 +114,12 @@ func TestTableLifecycleAndValidation(t *testing.T) {
 	ctx := context.Background()
 	mustKeyspace(t, m, "app")
 
-	// Missing keyspace rejected.
+	// Missing keyspace is ResourceNotFound (the documented "tried to access a
+	// keyspace ... that doesn't exist" error), not a validation failure.
 	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
 		KeyspaceName: "ghost", Name: "t", SchemaDefinition: sampleSchema(),
-	}); !cerrors.IsInvalidArgument(err) {
-		t.Fatalf("table in missing keyspace: got %v, want InvalidArgument", err)
+	}); !cerrors.IsNotFound(err) {
+		t.Fatalf("table in missing keyspace: got %v, want NotFound", err)
 	}
 
 	// No partition key rejected.
diff --git a/providers/aws/keyspaces/names_test.go b/providers/aws/keyspaces/names_test.go
new file mode 100644
index 000000000..03604fb1c
--- /dev/null
+++ b/providers/aws/keyspaces/names_test.go
@@ -0,0 +1,115 @@
+package keyspaces
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	cerrors "github.com/stackshy/cloudemu/v2/errors"
+	ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver"
+)
+
+// Keyspace/table names follow the documented KeyspaceName/TableName shape:
+// length 1-48, pattern [a-zA-Z0-9][a-zA-Z0-9_]{0,47}.
+func TestNameValidation(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+
+	maxName := strings.Repeat("a", maxNameLen)
+	tooLong := strings.Repeat("a", maxNameLen+1)
+
+	valid := []string{"a", "A1", "1abc", "9", "my_keyspace", maxName}
+	for _, n := range valid {
+		if _, err := m.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{Name: n}); err != nil {
+			t.Fatalf("CreateKeyspace(%q): unexpected error %v", n, err)
+		}
+	}
+
+	invalid := []string{"", tooLong, "_lead", "has-dash", "has.dot", "has space", "a/b", "ünicode"}
+	for _, n := range invalid {
+		if _, err := m.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{Name: n}); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("CreateKeyspace(%q): got %v, want InvalidArgument", n, err)
+		}
+
+		if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+			KeyspaceName: "a", Name: n, SchemaDefinition: sampleSchema(),
+		}); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("CreateTable(%q): got %v, want InvalidArgument", n, err)
+		}
+	}
+
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "a", Name: "1t", SchemaDefinition: sampleSchema(),
+	}); err != nil {
+		t.Fatalf("CreateTable leading digit: %v", err)
+	}
+
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "a", Name: maxName, SchemaDefinition: sampleSchema(),
+	}); err != nil {
+		t.Fatalf("CreateTable 48-char name: %v", err)
+	}
+
+	if _, err := m.RestoreTable(ctx, ksdriver.RestoreTableConfig{
+		SourceKeyspace: "a", SourceTable: "1t", TargetKeyspace: "a", TargetTable: tooLong,
+		RestoreTimestamp: m.opts.Clock.Now().Add(-1),
+	}); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("RestoreTable too-long target: got %v, want InvalidArgument", err)
+	}
+
+	fields := []ksdriver.FieldDefinition{{Name: "f", Type: "text"}}
+	if _, err := m.CreateType(ctx, "a", tooLong, fields); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("CreateType too long: got %v, want InvalidArgument", err)
+	}
+
+	if _, err := m.CreateType(ctx, "a", maxName, fields); err != nil {
+		t.Fatalf("CreateType 48-char name: %v", err)
+	}
+}
+
+// A well-formed request naming a parent keyspace that doesn't exist is
+// ResourceNotFound for CreateTable, CreateType, and RestoreTable's target.
+func TestMissingParentKeyspaceIsNotFound(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+	mustKeyspace(t, m, "app")
+
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "ghost", Name: "t", SchemaDefinition: sampleSchema(),
+	}); !cerrors.IsNotFound(err) {
+		t.Fatalf("CreateTable: got %v, want NotFound", err)
+	}
+
+	if _, err := m.CreateType(ctx, "ghost", "addr", []ksdriver.FieldDefinition{{Name: "f", Type: "text"}}); !cerrors.IsNotFound(err) {
+		t.Fatalf("CreateType: got %v, want NotFound", err)
+	}
+
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "app", Name: "src", SchemaDefinition: sampleSchema(),
+	}); err != nil {
+		t.Fatalf("CreateTable: %v", err)
+	}
+
+	if _, err := m.RestoreTable(ctx, ksdriver.RestoreTableConfig{
+		SourceKeyspace: "app", SourceTable: "src", TargetKeyspace: "ghost", TargetTable: "dst",
+		RestoreTimestamp: m.opts.Clock.Now(),
+	}); !cerrors.IsNotFound(err) {
+		t.Fatalf("RestoreTable: got %v, want NotFound", err)
+	}
+}
+
+// All four documented system keyspaces exist and none can be deleted.
+func TestSystemKeyspaces(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+
+	for _, name := range []string{"system", "system_schema", "system_schema_mcs", "system_multiregion_info"} {
+		if _, err := m.GetKeyspace(ctx, name); err != nil {
+			t.Fatalf("GetKeyspace(%s): %v", name, err)
+		}
+
+		if err := m.DeleteKeyspace(ctx, name); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("DeleteKeyspace(%s): got %v, want InvalidArgument", name, err)
+		}
+	}
+}
diff --git a/providers/aws/keyspaces/snapshot.go b/providers/aws/keyspaces/snapshot.go
index f57a670aa..9410446c8 100644
--- a/providers/aws/keyspaces/snapshot.go
+++ b/providers/aws/keyspaces/snapshot.go
@@ -47,9 +47,11 @@ func (m *Mock) Snapshot(_ context.Context, _ bool) (json.RawMessage, error) {
 		*d.dst = b
 	}
 
+	// Copy the tag maps under the lock: json.Marshal runs after the unlock, and
+	// TagResource mutates the inner maps in place.
 	m.mu.RLock()
 	if len(m.tags) > 0 {
-		snap.Tags = m.tags
+		snap.Tags = copyTagMaps(m.tags)
 	}
 	m.mu.RUnlock()
 
@@ -86,9 +88,58 @@ func (m *Mock) Restore(_ context.Context, data json.RawMessage) error {
 
 	if snap.Tags != nil {
 		m.mu.Lock()
-		m.tags = snap.Tags
+		m.tags = m.canonicalTagsLocked(snap.Tags)
 		m.mu.Unlock()
 	}
 
 	return nil
 }
+
+// canonicalTagsLocked keys restored tags by the canonical ARN of the keyspace
+// or table they belong to. Tag reads and writes resolve to the canonical ARN,
+// but an older build stored tags under whatever ARN the caller passed (for
+// example the slash-less table form) and accepted ARNs of resources that don't
+// exist. A non-canonical entry for an existing resource is merged into the
+// canonical one, with the canonical entry's values winning on a key clash. An
+// entry that names no existing resource is dropped: no API can reach it, and
+// keeping it would hand its tags to a resource created later under that ARN.
+// The caller holds the write lock, and the stores are already restored.
+func (m *Mock) canonicalTagsLocked(in map[string]map[string]string) map[string]map[string]string {
+	out := make(map[string]map[string]string, len(in))
+
+	for arn, tags := range in {
+		if m.existingResourceARN(arn) == arn {
+			out[arn] = copyTags(tags)
+		}
+	}
+
+	for arn, tags := range in {
+		canonical := m.existingResourceARN(arn)
+		if canonical == "" || canonical == arn || arn+"/" != canonical {
+			continue
+		}
+
+		cur := out[canonical]
+		if cur == nil {
+			cur = make(map[string]string, len(tags))
+			out[canonical] = cur
+		}
+
+		for k, v := range tags {
+			if _, ok := cur[k]; !ok {
+				cur[k] = v
+			}
+		}
+	}
+
+	return out
+}
+
+func copyTagMaps(in map[string]map[string]string) map[string]map[string]string {
+	out := make(map[string]map[string]string, len(in))
+	for arn, tags := range in {
+		out[arn] = copyTags(tags)
+	}
+
+	return out
+}
diff --git a/providers/aws/keyspaces/snapshot_test.go b/providers/aws/keyspaces/snapshot_test.go
index d626540f6..0e9eb712a 100644
--- a/providers/aws/keyspaces/snapshot_test.go
+++ b/providers/aws/keyspaces/snapshot_test.go
@@ -2,6 +2,10 @@ package keyspaces
 
 import (
 	"context"
+	"encoding/json"
+	"reflect"
+	"strconv"
+	"strings"
 	"testing"
 
 	ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver"
@@ -62,3 +66,111 @@ func TestSnapshotRoundTripKeyspaces(t *testing.T) {
 		t.Fatalf("restored tags = %+v", tags)
 	}
 }
+
+// Snapshot must not hand json.Marshal the live tag maps: TagResource mutates
+// them in place, so a concurrent tag write raced the marshal (go test -race).
+func TestSnapshotConcurrentWithTagResource(t *testing.T) {
+	ctx := context.Background()
+	m := newTestMock()
+	mustKeyspace(t, m, "app")
+
+	arn := m.keyspaceARN("app")
+	if err := m.TagResource(ctx, arn, map[string]string{"env": "prod"}); err != nil {
+		t.Fatalf("TagResource: %v", err)
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+
+		for i := 0; i < 500; i++ {
+			_ = m.TagResource(ctx, arn, map[string]string{"k" + strconv.Itoa(i%50): "v"})
+		}
+	}()
+
+	for i := 0; i < 200; i++ {
+		if _, err := m.Snapshot(ctx, false); err != nil {
+			t.Fatalf("Snapshot: %v", err)
+		}
+	}
+
+	<-done
+}
+
+// A snapshot written by an older build can hold tags under a non-canonical ARN
+// (the slash-less table form) or under an ARN that names no resource. Restore
+// moves the first onto the canonical ARN, so the tags stay reachable and are
+// dropped with the table, and discards the second, so a resource created later
+// under that ARN doesn't inherit them.
+func TestRestoreCanonicalizesLegacyTagARNs(t *testing.T) {
+	ctx := context.Background()
+	src := newTestMock()
+	mustKeyspace(t, src, "app")
+
+	if _, err := src.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "app", Name: "t1", SchemaDefinition: sampleSchema(),
+	}); err != nil {
+		t.Fatalf("CreateTable: %v", err)
+	}
+
+	tableARN := src.tableARN("app", "t1")
+	ghostARN := src.keyspaceARN("ghost")
+
+	raw, err := src.Snapshot(ctx, false)
+	if err != nil {
+		t.Fatalf("Snapshot: %v", err)
+	}
+
+	var snap keyspacesSnapshot
+	if err := json.Unmarshal(raw, &snap); err != nil {
+		t.Fatalf("decode snapshot: %v", err)
+	}
+
+	snap.Tags = map[string]map[string]string{
+		tableARN:                             {"env": "prod"},
+		strings.TrimSuffix(tableARN, "/"):    {"env": "old", "team": "data"},
+		ghostARN:                             {"stale": "yes"},
+		src.keyspaceARN("app") + "extra/x/y": {"junk": "1"},
+	}
+
+	legacy, err := json.Marshal(snap)
+	if err != nil {
+		t.Fatalf("encode snapshot: %v", err)
+	}
+
+	dst := newTestMock()
+	if err := dst.Restore(ctx, legacy); err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+
+	got, err := dst.ListTagsForResource(ctx, tableARN)
+	if err != nil {
+		t.Fatalf("ListTagsForResource: %v", err)
+	}
+
+	want := []ksdriver.Tag{{Key: "env", Value: "prod"}, {Key: "team", Value: "data"}}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("restored table tags = %+v, want %+v", got, want)
+	}
+
+	if len(dst.tags) != 1 {
+		t.Fatalf("restored tag store = %+v, want only the canonical table entry", dst.tags)
+	}
+
+	if _, err := dst.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{Name: "ghost"}); err != nil {
+		t.Fatalf("CreateKeyspace ghost: %v", err)
+	}
+
+	if tags, err := dst.ListTagsForResource(ctx, ghostARN); err != nil || len(tags) != 0 {
+		t.Fatalf("new keyspace inherited stale tags: %+v, %v", tags, err)
+	}
+
+	if err := dst.DeleteTable(ctx, "app", "t1"); err != nil {
+		t.Fatalf("DeleteTable: %v", err)
+	}
+
+	if len(dst.tags) != 0 {
+		t.Fatalf("tags left after DeleteTable: %+v", dst.tags)
+	}
+}
diff --git a/providers/aws/keyspaces/tables.go b/providers/aws/keyspaces/tables.go
index f4797578d..c97de5e36 100644
--- a/providers/aws/keyspaces/tables.go
+++ b/providers/aws/keyspaces/tables.go
@@ -117,6 +117,10 @@ func (m *Mock) CreateTable(_ context.Context, cfg ksdriver.CreateTableConfig) (*
 		return nil, err
 	}
 
+	if err := validateTags(cfg.Tags); err != nil {
+		return nil, err
+	}
+
 	if err := validateSchema(&cfg.SchemaDefinition); err != nil {
 		return nil, err
 	}
@@ -125,7 +129,7 @@ func (m *Mock) CreateTable(_ context.Context, cfg ksdriver.CreateTableConfig) (*
 	defer m.mu.Unlock()
 
 	if !m.keyspaces.Has(cfg.KeyspaceName) {
-		return nil, cerrors.Newf(cerrors.InvalidArgument, "keyspace %q not found", cfg.KeyspaceName)
+		return nil, cerrors.Newf(cerrors.NotFound, "keyspace %q not found", cfg.KeyspaceName)
 	}
 
 	key := tableKey(cfg.KeyspaceName, cfg.Name)
@@ -346,6 +350,10 @@ func (m *Mock) RestoreTable(_ context.Context, cfg ksdriver.RestoreTableConfig)
 		return nil, err
 	}
 
+	if err := validateTags(cfg.Tags); err != nil {
+		return nil, err
+	}
+
 	if cfg.RestoreTimestamp.IsZero() {
 		return nil, cerrors.New(cerrors.InvalidArgument, "restoreTimestamp is required")
 	}
@@ -363,7 +371,7 @@ func (m *Mock) RestoreTable(_ context.Context, cfg ksdriver.RestoreTableConfig)
 	}
 
 	if !m.keyspaces.Has(cfg.TargetKeyspace) {
-		return nil, cerrors.Newf(cerrors.InvalidArgument, "target keyspace %q not found", cfg.TargetKeyspace)
+		return nil, cerrors.Newf(cerrors.NotFound, "target keyspace %q not found", cfg.TargetKeyspace)
 	}
 
 	targetKey := tableKey(cfg.TargetKeyspace, cfg.TargetTable)
diff --git a/providers/aws/keyspaces/tags.go b/providers/aws/keyspaces/tags.go
index 27ec4869c..531727c20 100644
--- a/providers/aws/keyspaces/tags.go
+++ b/providers/aws/keyspaces/tags.go
@@ -2,7 +2,10 @@ package keyspaces
 
 import (
 	"context"
+	"regexp"
 	"sort"
+	"strings"
+	"unicode/utf8"
 
 	cerrors "github.com/stackshy/cloudemu/v2/errors"
 	ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver"
@@ -26,26 +29,164 @@ func (m *Mock) setTags(arn string, tags map[string]string) {
 	m.tags[arn] = cur
 }
 
-// TagResource adds tags to a resource by ARN.
-func (m *Mock) TagResource(_ context.Context, arn string, tags map[string]string) error {
+// resourceARNPattern is the documented resourceArn pattern for the tagging
+// operations (arn:(aws[a-zA-Z0-9-]*):cassandra:.+.*, length 20-1000).
+var resourceARNPattern = regexp.MustCompile(`^arn:aws[a-zA-Z0-9-]*:cassandra:.+`)
+
+// tagCharsPattern is the documented tag key/value character set: letters,
+// white space, numbers, and the characters + - = . _ : and /.
+var tagCharsPattern = regexp.MustCompile(`^[\p{L}\p{Z}\p{N}+\-=._:/]+$`)
+
+const (
+	minResourceARNLen = 20
+	maxResourceARNLen = 1000
+	maxTagKeyLen      = 128
+	maxTagValueLen    = 256
+	// maxTagListLen is the documented TagList maximum (Array Members: Maximum
+	// number of 60 items) on TagResource, UntagResource, CreateKeyspace,
+	// CreateTable and RestoreTable.
+	maxTagListLen     = 60
+	reservedTagPrefix = "aws:"
+	arnResourceMarker = ":/keyspace/"
+	arnTableSegment   = "table"
+)
+
+// validateTags applies the documented Amazon Keyspaces tag restrictions: key
+// 1-128 and value 1-256 Unicode characters (a value can't be empty), only
+// letters, white space, numbers and + - = . _ : /, and no user-assigned keys
+// with the reserved "aws:" prefix. A request carries at most 60 tags.
+func validateTags(tags map[string]string) error {
+	if len(tags) > maxTagListLen {
+		return cerrors.Newf(cerrors.InvalidArgument, "tags must contain at most %d items", maxTagListLen)
+	}
+
+	for k, v := range tags {
+		if n := utf8.RuneCountInString(k); n == 0 || n > maxTagKeyLen {
+			return cerrors.Newf(cerrors.InvalidArgument, "tag key %q must be 1-%d characters", k, maxTagKeyLen)
+		}
+
+		if n := utf8.RuneCountInString(v); n == 0 || n > maxTagValueLen {
+			return cerrors.Newf(cerrors.InvalidArgument,
+				"tag value for key %q must be 1-%d characters", k, maxTagValueLen)
+		}
+
+		if strings.HasPrefix(strings.ToLower(k), reservedTagPrefix) {
+			return cerrors.Newf(cerrors.InvalidArgument, "tag key %q uses the reserved prefix %q", k, reservedTagPrefix)
+		}
+
+		if !tagCharsPattern.MatchString(k) || !tagCharsPattern.MatchString(v) {
+			return cerrors.Newf(cerrors.InvalidArgument,
+				"tag %q contains characters outside letters, numbers, white space and + - = . _ : /", k)
+		}
+	}
+
+	return nil
+}
+
+// resolveTaggableLocked validates a resource ARN and returns the canonical ARN
+// of the existing keyspace or table it names. Real Keyspaces rejects a
+// malformed ARN with ValidationException and an unknown resource with
+// ResourceNotFoundException. The trailing "/" is optional, so both the form
+// this emulator returns and the slash-less table form in the AWS docs resolve
+// to the same resource. The caller holds the lock.
+func (m *Mock) resolveTaggableLocked(arn string) (string, error) {
+	if err := validateResourceARN(arn); err != nil {
+		return "", err
+	}
+
+	canonical := m.existingResourceARN(arn)
+
+	// The region/account (and partition) must match this emulator's too.
+	if canonical == "" || (arn != canonical && arn+"/" != canonical) {
+		return "", cerrors.Newf(cerrors.NotFound, "resource %q not found", arn)
+	}
+
+	return canonical, nil
+}
+
+// validateResourceARN checks the documented resourceArn length and pattern.
+func validateResourceARN(arn string) error {
 	if arn == "" {
 		return cerrors.New(cerrors.InvalidArgument, "resourceArn is required")
 	}
 
+	if len(arn) < minResourceARNLen || len(arn) > maxResourceARNLen || !resourceARNPattern.MatchString(arn) {
+		return cerrors.Newf(cerrors.InvalidArgument, "resourceArn %q is not a valid Amazon Keyspaces ARN", arn)
+	}
+
+	return nil
+}
+
+// existingResourceARN returns the canonical ARN of the keyspace
+// (".../keyspace/<ks>/") or table (".../keyspace/<ks>/table/<t>/") that arn's
+// resource path names, or "" when the path names no existing keyspace or
+// table. The caller holds the lock.
+func (m *Mock) existingResourceARN(arn string) string {
+	idx := strings.Index(arn, arnResourceMarker)
+	if idx < 0 {
+		return ""
+	}
+
+	parts := strings.Split(strings.TrimSuffix(arn[idx+len(arnResourceMarker):], "/"), "/")
+
+	switch {
+	case len(parts) == 1 && m.keyspaces.Has(parts[0]):
+		return m.keyspaceARN(parts[0])
+	case len(parts) == 3 && parts[1] == arnTableSegment && m.tables.Has(tableKey(parts[0], parts[2])):
+		return m.tableARN(parts[0], parts[2])
+	default:
+		return ""
+	}
+}
+
+// TagResource adds tags to an existing keyspace or table by ARN.
+func (m *Mock) TagResource(_ context.Context, arn string, tags map[string]string) error {
+	if len(tags) == 0 {
+		return cerrors.New(cerrors.InvalidArgument, "tags must contain at least one tag")
+	}
+
+	if err := validateTags(tags); err != nil {
+		return err
+	}
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
-	m.setTags(arn, tags)
+	canonical, err := m.resolveTaggableLocked(arn)
+	if err != nil {
+		return err
+	}
+
+	m.setTags(canonical, tags)
 
 	return nil
 }
 
-// UntagResource removes tag keys from a resource by ARN.
+// UntagResource removes tag keys from an existing keyspace or table by ARN.
 func (m *Mock) UntagResource(_ context.Context, arn string, keys []string) error {
+	if len(keys) == 0 {
+		return cerrors.New(cerrors.InvalidArgument, "tags must contain at least one tag")
+	}
+
+	if len(keys) > maxTagListLen {
+		return cerrors.Newf(cerrors.InvalidArgument, "tags must contain at most %d items", maxTagListLen)
+	}
+
+	for _, k := range keys {
+		if n := utf8.RuneCountInString(k); n == 0 || n > maxTagKeyLen {
+			return cerrors.Newf(cerrors.InvalidArgument, "tag key %q must be 1-%d characters", k, maxTagKeyLen)
+		}
+	}
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
-	cur := m.tags[arn]
+	canonical, err := m.resolveTaggableLocked(arn)
+	if err != nil {
+		return err
+	}
+
+	cur := m.tags[canonical]
 	for _, k := range keys {
 		delete(cur, k)
 	}
@@ -53,13 +194,18 @@ func (m *Mock) UntagResource(_ context.Context, arn string, keys []string) error
 	return nil
 }
 
-// ListTagsForResource returns a resource's tags in deterministic (sorted)
-// order (AWS leaves tag order unspecified).
+// ListTagsForResource returns an existing resource's tags in deterministic
+// (sorted) order (AWS leaves tag order unspecified).
 func (m *Mock) ListTagsForResource(_ context.Context, arn string) ([]ksdriver.Tag, error) {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
 
-	tags := m.tags[arn]
+	canonical, err := m.resolveTaggableLocked(arn)
+	if err != nil {
+		return nil, err
+	}
+
+	tags := m.tags[canonical]
 
 	names := make([]string, 0, len(tags))
 	for k := range tags {
diff --git a/providers/aws/keyspaces/tags_test.go b/providers/aws/keyspaces/tags_test.go
new file mode 100644
index 000000000..7c7dbe9b6
--- /dev/null
+++ b/providers/aws/keyspaces/tags_test.go
@@ -0,0 +1,223 @@
+package keyspaces
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	cerrors "github.com/stackshy/cloudemu/v2/errors"
+	ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver"
+)
+
+// Tagging operations require an existing keyspace or table and a well-formed
+// Keyspaces ARN.
+func TestTagsRequireExistingResource(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+	mustKeyspace(t, m, "app")
+
+	ghostKS := m.keyspaceARN("ghost")
+	ghostTable := m.tableARN("app", "ghost")
+
+	for _, arn := range []string{ghostKS, ghostTable} {
+		if err := m.TagResource(ctx, arn, map[string]string{"k": "v"}); !cerrors.IsNotFound(err) {
+			t.Fatalf("TagResource(%s): got %v, want NotFound", arn, err)
+		}
+
+		if err := m.UntagResource(ctx, arn, []string{"k"}); !cerrors.IsNotFound(err) {
+			t.Fatalf("UntagResource(%s): got %v, want NotFound", arn, err)
+		}
+
+		if _, err := m.ListTagsForResource(ctx, arn); !cerrors.IsNotFound(err) {
+			t.Fatalf("ListTagsForResource(%s): got %v, want NotFound", arn, err)
+		}
+	}
+
+	if len(m.tags) != 0 {
+		t.Fatalf("tags stored for missing resources: %v", m.tags)
+	}
+
+	for _, arn := range []string{"", "not-an-arn", "arn:aws:dynamodb:us-east-1:123456789012:table/x"} {
+		if err := m.TagResource(ctx, arn, map[string]string{"k": "v"}); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("TagResource(%q): got %v, want InvalidArgument", arn, err)
+		}
+	}
+
+	// Existing keyspace and table ARNs still work, including after deletion
+	// turning them into NotFound.
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "app", Name: "t", SchemaDefinition: sampleSchema(),
+	}); err != nil {
+		t.Fatalf("CreateTable: %v", err)
+	}
+
+	tableARN := m.tableARN("app", "t")
+	for _, arn := range []string{m.keyspaceARN("app"), tableARN} {
+		if err := m.TagResource(ctx, arn, map[string]string{"k": "v"}); err != nil {
+			t.Fatalf("TagResource(%s): %v", arn, err)
+		}
+
+		if tags, err := m.ListTagsForResource(ctx, arn); err != nil || len(tags) != 1 {
+			t.Fatalf("ListTagsForResource(%s): %v %v", arn, tags, err)
+		}
+	}
+
+	if err := m.DeleteTable(ctx, "app", "t"); err != nil {
+		t.Fatalf("DeleteTable: %v", err)
+	}
+
+	if _, err := m.ListTagsForResource(ctx, tableARN); !cerrors.IsNotFound(err) {
+		t.Fatalf("ListTagsForResource after delete: got %v, want NotFound", err)
+	}
+}
+
+// Tag keys/values follow the documented Keyspaces tag restrictions on every
+// path that accepts tags.
+func TestTagRestrictions(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+	mustKeyspace(t, m, "app")
+
+	arn := m.keyspaceARN("app")
+
+	bad := []map[string]string{
+		{"aws:owner": "x"},
+		{"AWS:owner": "x"},
+		{"": "x"},
+		{"k": ""},
+		{strings.Repeat("k", maxTagKeyLen+1): "v"},
+		{"k": strings.Repeat("v", maxTagValueLen+1)},
+		{"k*": "v"},
+		{"k": "v@x"},
+	}
+
+	for _, tags := range bad {
+		if err := m.TagResource(ctx, arn, tags); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("TagResource(%v): got %v, want InvalidArgument", tags, err)
+		}
+
+		if _, err := m.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{Name: "k2", Tags: tags}); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("CreateKeyspace tags %v: got %v, want InvalidArgument", tags, err)
+		}
+
+		if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+			KeyspaceName: "app", Name: "t2", SchemaDefinition: sampleSchema(), Tags: tags,
+		}); !cerrors.IsInvalidArgument(err) {
+			t.Fatalf("CreateTable tags %v: got %v, want InvalidArgument", tags, err)
+		}
+	}
+
+	if err := m.TagResource(ctx, arn, nil); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("TagResource no tags: got %v, want InvalidArgument", err)
+	}
+
+	if err := m.UntagResource(ctx, arn, nil); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("UntagResource no tags: got %v, want InvalidArgument", err)
+	}
+
+	good := map[string]string{
+		strings.Repeat("k", maxTagKeyLen): strings.Repeat("v", maxTagValueLen),
+		"team name":                       "data + ml = a/b_c:d.e-f",
+		"ключ":                            "значение1",
+		"awsish":                          "ok",
+	}
+	if err := m.TagResource(ctx, arn, good); err != nil {
+		t.Fatalf("TagResource valid: %v", err)
+	}
+
+	if tags, err := m.ListTagsForResource(ctx, arn); err != nil || len(tags) != len(good) {
+		t.Fatalf("ListTagsForResource: %v %v", tags, err)
+	}
+}
+
+// The documented TagList bound (at most 60 items per request) applies to the
+// library path too, not only to the wire handler, on every operation that
+// takes tags. Exactly 60 is accepted.
+func TestTagListBound(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+	mustKeyspace(t, m, "app")
+
+	arn := m.keyspaceARN("app")
+
+	tooMany := make(map[string]string, maxTagListLen+1)
+	keys := make([]string, 0, maxTagListLen+1)
+
+	for i := 0; i <= maxTagListLen; i++ {
+		k := "k" + strconv.Itoa(i)
+		tooMany[k] = "v"
+		keys = append(keys, k)
+	}
+
+	if err := m.TagResource(ctx, arn, tooMany); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("TagResource 61 tags: got %v, want InvalidArgument", err)
+	}
+
+	if err := m.UntagResource(ctx, arn, keys); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("UntagResource 61 keys: got %v, want InvalidArgument", err)
+	}
+
+	if _, err := m.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{Name: "k2", Tags: tooMany}); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("CreateKeyspace 61 tags: got %v, want InvalidArgument", err)
+	}
+
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "app", Name: "t2", SchemaDefinition: sampleSchema(), Tags: tooMany,
+	}); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("CreateTable 61 tags: got %v, want InvalidArgument", err)
+	}
+
+	if _, err := m.RestoreTable(ctx, ksdriver.RestoreTableConfig{
+		SourceKeyspace: "app", SourceTable: "t1", TargetKeyspace: "app", TargetTable: "t3",
+		RestoreTimestamp: time.Now(), Tags: tooMany,
+	}); !cerrors.IsInvalidArgument(err) {
+		t.Fatalf("RestoreTable 61 tags: got %v, want InvalidArgument", err)
+	}
+
+	delete(tooMany, "k0")
+
+	if err := m.TagResource(ctx, arn, tooMany); err != nil {
+		t.Fatalf("TagResource 60 tags: %v", err)
+	}
+
+	if err := m.UntagResource(ctx, arn, keys[1:]); err != nil {
+		t.Fatalf("UntagResource 60 keys: %v", err)
+	}
+}
+
+// The trailing "/" on a resource ARN is optional; both forms address the same
+// resource, and an ARN for another account or region is not found.
+func TestTagARNForms(t *testing.T) {
+	m := newTestMock()
+	ctx := context.Background()
+	mustKeyspace(t, m, "app")
+
+	if _, err := m.CreateTable(ctx, ksdriver.CreateTableConfig{
+		KeyspaceName: "app", Name: "t", SchemaDefinition: sampleSchema(),
+	}); err != nil {
+		t.Fatalf("CreateTable: %v", err)
+	}
+
+	noSlash := strings.TrimSuffix(m.tableARN("app", "t"), "/")
+	if err := m.TagResource(ctx, noSlash, map[string]string{"k": "v"}); err != nil {
+		t.Fatalf("TagResource without trailing slash: %v", err)
+	}
+
+	tags, err := m.ListTagsForResource(ctx, m.tableARN("app", "t"))
+	if err != nil || len(tags) != 1 {
+		t.Fatalf("ListTagsForResource canonical: %v %v", tags, err)
+	}
+
+	for _, arn := range []string{
+		"arn:aws:cassandra:us-west-2:123456789012:/keyspace/app/",
+		"arn:aws:cassandra:us-east-1:999999999999:/keyspace/app/",
+		"arn:aws:cassandra:us-east-1:123456789012:/keyspace/app/table/",
+		"arn:aws:cassandra:us-east-1:123456789012:/keyspace/app/type/t/",
+	} {
+		if _, err := m.ListTagsForResource(ctx, arn); !cerrors.IsNotFound(err) {
+			t.Fatalf("ListTagsForResource(%s): got %v, want NotFound", arn, err)
+		}
+	}
+}
diff --git a/providers/aws/keyspaces/udt.go b/providers/aws/keyspaces/udt.go
index 5f32f0e98..6bd9365a1 100644
--- a/providers/aws/keyspaces/udt.go
+++ b/providers/aws/keyspaces/udt.go
@@ -101,7 +101,7 @@ func cloneUDT(in *ksdriver.UDT) ksdriver.UDT {
 
 // CreateType creates a user-defined type in an existing keyspace.
 func (m *Mock) CreateType(_ context.Context, keyspace, name string, fields []ksdriver.FieldDefinition) (*ksdriver.UDT, error) {
-	if err := validName("type", name); err != nil {
+	if err := validTypeName(name); err != nil {
 		return nil, err
 	}
 
@@ -113,7 +113,7 @@ func (m *Mock) CreateType(_ context.Context, keyspace, name string, fields []ksd
 	defer m.mu.Unlock()
 
 	if !m.keyspaces.Has(keyspace) {
-		return nil, cerrors.Newf(cerrors.InvalidArgument, "keyspace %q not found", keyspace)
+		return nil, cerrors.Newf(cerrors.NotFound, "keyspace %q not found", keyspace)
 	}
 
 	key := typeKey(keyspace, name)
diff --git a/providers/aws/rgt_discovery.go b/providers/aws/rgt_discovery.go
index 2c263bba6..cd2d207e4 100644
--- a/providers/aws/rgt_discovery.go
+++ b/providers/aws/rgt_discovery.go
@@ -3,6 +3,7 @@ package aws
 import (
 	"context"
 
+	cerrors "github.com/stackshy/cloudemu/v2/errors"
 	"github.com/stackshy/cloudemu/v2/internal/idgen"
 	acmdriver "github.com/stackshy/cloudemu/v2/services/acm/driver"
 	cloudtraildriver "github.com/stackshy/cloudemu/v2/services/cloudtrail/driver"
@@ -419,6 +420,12 @@ func (d taggedDiscovery) discoverConfig(ctx context.Context) ([]resourcediscover
 // collected rows. It factors out the identical enumerate → read-tags → row loop
 // the single-list discoverers share. The projection returns its error last so a
 // tag-read failure inside fn short-circuits the walk with that error.
+//
+// A NotFound from fn means the resource was deleted between the list call and
+// its tag read. It is skipped, the same as a resource that was never listed,
+// so a delete racing GetResources doesn't fail the whole inventory (the
+// resourcediscovery walkers handle this race the same way). Any other error
+// still propagates.
 func project[T any](ctx context.Context, items []T,
 	fn func(context.Context, T) (resourcediscovery.DiscoveredResource, error),
 ) ([]resourcediscovery.DiscoveredResource, error) {
@@ -426,6 +433,10 @@ func project[T any](ctx context.Context, items []T,
 
 	for i := range items {
 		r, err := fn(ctx, items[i])
+		if cerrors.IsNotFound(err) {
+			continue
+		}
+
 		if err != nil {
 			return nil, err
 		}
diff --git a/providers/aws/rgt_discovery_test.go b/providers/aws/rgt_discovery_test.go
new file mode 100644
index 000000000..3a1255891
--- /dev/null
+++ b/providers/aws/rgt_discovery_test.go
@@ -0,0 +1,105 @@
+package aws
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+
+	cerrors "github.com/stackshy/cloudemu/v2/errors"
+	ksdriver "github.com/stackshy/cloudemu/v2/services/keyspaces/driver"
+	"github.com/stackshy/cloudemu/v2/services/resourcediscovery"
+)
+
+// A resource deleted between the list call and its tag read makes the tag read
+// return NotFound. project must drop that resource instead of failing, since
+// the error would otherwise abort DiscoverResources for every service. Any
+// other error must still propagate.
+func TestProjectSkipsVanishedResource(t *testing.T) {
+	ctx := context.Background()
+	boom := errors.New("boom")
+
+	read := func(_ context.Context, name string) (resourcediscovery.DiscoveredResource, error) {
+		switch name {
+		case "gone":
+			return resourcediscovery.DiscoveredResource{}, cerrors.Newf(cerrors.NotFound, "resource %q not found", name)
+		case "broken":
+			return resourcediscovery.DiscoveredResource{}, boom
+		default:
+			return row("svc", "thing", name, "arn:"+name, nil), nil
+		}
+	}
+
+	rows, err := project(ctx, []string{"a", "gone", "b"}, read)
+	if err != nil {
+		t.Fatalf("project with a vanished resource: %v", err)
+	}
+
+	if len(rows) != 2 || rows[0].ID != "a" || rows[1].ID != "b" {
+		t.Fatalf("rows = %+v, want a and b only", rows)
+	}
+
+	if _, err := project(ctx, []string{"a", "broken"}, read); !errors.Is(err, boom) {
+		t.Fatalf("project with a failing read: err = %v, want boom", err)
+	}
+}
+
+// A DeleteKeyspace racing the Keyspaces discovery walk must not fail it.
+// ListTagsForResource returns NotFound for a keyspace that is gone, so without
+// the skip in project nearly every walk that overlaps the delete loop fails.
+func TestKeyspacesDiscoverySurvivesConcurrentDelete(t *testing.T) {
+	ctx := context.Background()
+	p := New()
+	d := taggedDiscovery{p}
+
+	if _, err := p.Keyspaces.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{
+		Name: "stable", Tags: map[string]string{"env": "prod"},
+	}); err != nil {
+		t.Fatalf("CreateKeyspace stable: %v", err)
+	}
+
+	stop := make(chan struct{})
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+
+			_, _ = p.Keyspaces.CreateKeyspace(ctx, ksdriver.CreateKeyspaceConfig{Name: "churn"})
+			_ = p.Keyspaces.DeleteKeyspace(ctx, "churn")
+		}
+	}()
+
+	defer func() {
+		close(stop)
+		wg.Wait()
+	}()
+
+	for i := 0; i < 2000; i++ {
+		rows, err := d.discoverKeyspaces(ctx)
+		if err != nil {
+			t.Fatalf("walk %d: discoverKeyspaces: %v", i, err)
+		}
+
+		var found bool
+
+		for _, r := range rows {
+			if r.ID == "stable" && r.Tags["env"] == "prod" {
+				found = true
+			}
+		}
+
+		if !found {
+			t.Fatalf("walk %d: stable keyspace missing or untagged: %+v", i, rows)
+		}
+	}
+}
diff --git a/server/aws/keyspaces/keyspaces.go b/server/aws/keyspaces/keyspaces.go
index 35f4871bf..b1ed93707 100644
--- a/server/aws/keyspaces/keyspaces.go
+++ b/server/aws/keyspaces/keyspaces.go
@@ -17,6 +17,10 @@ func (h *Handler) createKeyspace(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if rejectTagListTooLong(w, len(in.Tags)) {
+		return
+	}
+
 	cfg := ksdriver.CreateKeyspaceConfig{Name: aws.ToString(in.KeyspaceName), Tags: tagMap(in.Tags)}
 	if in.ReplicationSpecification != nil {
 		cfg.ReplicationStrategy = string(in.ReplicationSpecification.ReplicationStrategy)
diff --git a/server/aws/keyspaces/sdk_roundtrip_test.go b/server/aws/keyspaces/sdk_roundtrip_test.go
index eaa054a9b..a257eeb4a 100644
--- a/server/aws/keyspaces/sdk_roundtrip_test.go
+++ b/server/aws/keyspaces/sdk_roundtrip_test.go
@@ -77,8 +77,8 @@ func TestSDKKeyspaceLifecycle(t *testing.T) {
 		t.Fatalf("ListKeyspaces: %v", err)
 	}
 
-	if len(list.Keyspaces) != 4 { // 3 system + app
-		t.Fatalf("expected 4 keyspaces, got %d", len(list.Keyspaces))
+	if len(list.Keyspaces) != 5 { // 4 system + app
+		t.Fatalf("expected 5 keyspaces, got %d", len(list.Keyspaces))
 	}
 
 	if _, err := client.DeleteKeyspace(ctx, &awskeyspaces.DeleteKeyspaceInput{KeyspaceName: aws.String("app")}); err != nil {
diff --git a/server/aws/keyspaces/tables.go b/server/aws/keyspaces/tables.go
index c981abdc4..7d80b49cb 100644
--- a/server/aws/keyspaces/tables.go
+++ b/server/aws/keyspaces/tables.go
@@ -60,6 +60,10 @@ func (h *Handler) createTable(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if rejectTagListTooLong(w, len(in.Tags)) {
+		return
+	}
+
 	cfg := ksdriver.CreateTableConfig{
 		KeyspaceName:            aws.ToString(in.KeyspaceName),
 		Name:                    aws.ToString(in.TableName),
@@ -248,6 +252,10 @@ func (h *Handler) restoreTable(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if rejectTagListTooLong(w, len(in.TagsOverride)) {
+		return
+	}
+
 	cfg := ksdriver.RestoreTableConfig{
 		SourceKeyspace:   aws.ToString(in.SourceKeyspaceName),
 		SourceTable:      aws.ToString(in.SourceTableName),
diff --git a/server/aws/keyspaces/tables_test.go b/server/aws/keyspaces/tables_test.go
new file mode 100644
index 000000000..4922733c9
--- /dev/null
+++ b/server/aws/keyspaces/tables_test.go
@@ -0,0 +1,120 @@
+package keyspaces_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awskeyspaces "github.com/aws/aws-sdk-go-v2/service/keyspaces"
+	kstypes "github.com/aws/aws-sdk-go-v2/service/keyspaces/types"
+
+	"github.com/stackshy/cloudemu/v2/config"
+	ksprovider "github.com/stackshy/cloudemu/v2/providers/aws/keyspaces"
+	ksserver "github.com/stackshy/cloudemu/v2/server/aws/keyspaces"
+)
+
+// Keyspace/table names must be 1-48 chars matching [a-zA-Z0-9][a-zA-Z0-9_]*;
+// a missing parent keyspace is ResourceNotFoundException.
+func TestSDKNameValidationAndMissingKeyspace(t *testing.T) {
+	client := newSDKClient(t)
+	ctx := context.Background()
+
+	var ve *kstypes.ValidationException
+
+	for _, n := range []string{strings.Repeat("a", 49), "bad-name", "_lead"} {
+		_, err := client.CreateKeyspace(ctx, &awskeyspaces.CreateKeyspaceInput{KeyspaceName: aws.String(n)})
+		if !errors.As(err, &ve) {
+			t.Fatalf("CreateKeyspace(%q): got %v, want ValidationException", n, err)
+		}
+	}
+
+	if _, err := client.CreateKeyspace(ctx, &awskeyspaces.CreateKeyspaceInput{KeyspaceName: aws.String("1app")}); err != nil {
+		t.Fatalf("CreateKeyspace leading digit: %v", err)
+	}
+
+	_, err := client.CreateTable(ctx, &awskeyspaces.CreateTableInput{
+		KeyspaceName: aws.String("1app"), TableName: aws.String(strings.Repeat("t", 49)), SchemaDefinition: sampleSchema(),
+	})
+	if !errors.As(err, &ve) {
+		t.Fatalf("CreateTable too-long name: got %v, want ValidationException", err)
+	}
+
+	var rnf *kstypes.ResourceNotFoundException
+
+	_, err = client.CreateTable(ctx, &awskeyspaces.CreateTableInput{
+		KeyspaceName: aws.String("ghost"), TableName: aws.String("t"), SchemaDefinition: sampleSchema(),
+	})
+	if !errors.As(err, &rnf) {
+		t.Fatalf("CreateTable missing keyspace: got %v, want ResourceNotFoundException", err)
+	}
+}
+
+// GetTable on a PAY_PER_REQUEST table without CDC omits readCapacityUnits,
+// writeCapacityUnits, and cdcSpecification.viewType; PROVISIONED keeps them.
+func TestGetTableOmitsUnsetFields(t *testing.T) {
+	opts := config.NewOptions(config.WithRegion("us-east-1"), config.WithAccountID("123456789012"))
+	h := ksserver.New(ksprovider.New(opts))
+
+	call := func(op, body string) map[string]any {
+		t.Helper()
+
+		req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
+		req.Header.Set("X-Amz-Target", "KeyspacesService."+op)
+		req.Header.Set("Content-Type", "application/x-amz-json-1.0")
+
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, req)
+
+		if rec.Code != http.StatusOK {
+			t.Fatalf("%s: status %d: %s", op, rec.Code, rec.Body.String())
+		}
+
+		var out map[string]any
+		if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
+			t.Fatalf("%s: decode: %v", op, err)
+		}
+
+		return out
+	}
+
+	schema := `"schemaDefinition":{"allColumns":[{"name":"id","type":"int"}],"partitionKeys":[{"name":"id"}]}`
+
+	call("CreateKeyspace", `{"keyspaceName":"app"}`)
+	call("CreateTable", `{"keyspaceName":"app","tableName":"ppr",`+schema+`}`)
+	call("CreateTable", `{"keyspaceName":"app","tableName":"prov",`+schema+
+		`,"capacitySpecification":{"throughputMode":"PROVISIONED","readCapacityUnits":5,"writeCapacityUnits":7}}`)
+
+	ppr := call("GetTable", `{"keyspaceName":"app","tableName":"ppr"}`)
+
+	capSpec, _ := ppr["capacitySpecification"].(map[string]any)
+	if capSpec["throughputMode"] != "PAY_PER_REQUEST" {
+		t.Fatalf("ppr throughputMode: %v", capSpec)
+	}
+
+	for _, k := range []string{"readCapacityUnits", "writeCapacityUnits"} {
+		if _, ok := capSpec[k]; ok {
+			t.Fatalf("ppr capacitySpecification has %s: %v", k, capSpec)
+		}
+	}
+
+	cdc, _ := ppr["cdcSpecification"].(map[string]any)
+	if cdc["status"] != "DISABLED" {
+		t.Fatalf("cdc status: %v", cdc)
+	}
+
+	if _, ok := cdc["viewType"]; ok {
+		t.Fatalf("cdcSpecification has empty viewType: %v", cdc)
+	}
+
+	prov := call("GetTable", `{"keyspaceName":"app","tableName":"prov"}`)
+
+	pcap, _ := prov["capacitySpecification"].(map[string]any)
+	if pcap["readCapacityUnits"] != float64(5) || pcap["writeCapacityUnits"] != float64(7) {
+		t.Fatalf("provisioned capacity: %v", pcap)
+	}
+}
diff --git a/server/aws/keyspaces/tags.go b/server/aws/keyspaces/tags.go
index 069aac061..8196b06bf 100644
--- a/server/aws/keyspaces/tags.go
+++ b/server/aws/keyspaces/tags.go
@@ -1,6 +1,7 @@
 package keyspaces
 
 import (
+	"fmt"
 	"net/http"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -15,6 +16,10 @@ func (h *Handler) tagResource(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if rejectTagListTooLong(w, len(in.Tags)) {
+		return
+	}
+
 	if err := h.db.TagResource(r.Context(), aws.ToString(in.ResourceArn), tagMap(in.Tags)); err != nil {
 		writeErr(w, err)
 		return
@@ -29,6 +34,10 @@ func (h *Handler) untagResource(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if rejectTagListTooLong(w, len(in.Tags)) {
+		return
+	}
+
 	keys := make([]string, 0, len(in.Tags))
 	for i := range in.Tags {
 		keys = append(keys, aws.ToString(in.Tags[i].Key))
@@ -62,3 +71,23 @@ func (h *Handler) listTagsForResource(w http.ResponseWriter, r *http.Request) {
 
 	writeJSON(w, keyspaces.ListTagsForResourceOutput{Tags: toWireTags(page), NextToken: next})
 }
+
+// maxTagListLen is the documented TagList maximum (Array Members: Maximum
+// number of 60 items) on TagResource, UntagResource, CreateKeyspace,
+// CreateTable and RestoreTable. The provider applies the same bound to the
+// tags it receives; this check runs on the raw wire list, which can repeat a
+// key and so be longer than the map the provider sees.
+const maxTagListLen = 60
+
+// rejectTagListTooLong writes a ValidationException and returns true when a
+// request's tag list exceeds the documented TagList size.
+func rejectTagListTooLong(w http.ResponseWriter, n int) bool {
+	if n <= maxTagListLen {
+		return false
+	}
+
+	wire.WriteJSONError(w, http.StatusBadRequest, "ValidationException",
+		fmt.Sprintf("tags must contain at most %d items", maxTagListLen))
+
+	return true
+}
diff --git a/server/aws/keyspaces/tags_test.go b/server/aws/keyspaces/tags_test.go
new file mode 100644
index 000000000..759b42d4b
--- /dev/null
+++ b/server/aws/keyspaces/tags_test.go
@@ -0,0 +1,100 @@
+package keyspaces_test
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awskeyspaces "github.com/aws/aws-sdk-go-v2/service/keyspaces"
+	kstypes "github.com/aws/aws-sdk-go-v2/service/keyspaces/types"
+)
+
+// Tagging a keyspace/table ARN that doesn't exist is ResourceNotFoundException.
+func TestSDKTagsOnMissingResource(t *testing.T) {
+	client := newSDKClient(t)
+	ctx := context.Background()
+
+	if _, err := client.CreateKeyspace(ctx, &awskeyspaces.CreateKeyspaceInput{KeyspaceName: aws.String("app")}); err != nil {
+		t.Fatalf("CreateKeyspace: %v", err)
+	}
+
+	ghost := aws.String("arn:aws:cassandra:us-east-1:123456789012:/keyspace/ghost/")
+	tags := []kstypes.Tag{{Key: aws.String("k"), Value: aws.String("v")}}
+
+	var rnf *kstypes.ResourceNotFoundException
+
+	_, err := client.TagResource(ctx, &awskeyspaces.TagResourceInput{ResourceArn: ghost, Tags: tags})
+	if !errors.As(err, &rnf) {
+		t.Fatalf("TagResource: got %v, want ResourceNotFoundException", err)
+	}
+
+	_, err = client.UntagResource(ctx, &awskeyspaces.UntagResourceInput{ResourceArn: ghost, Tags: tags})
+	if !errors.As(err, &rnf) {
+		t.Fatalf("UntagResource: got %v, want ResourceNotFoundException", err)
+	}
+
+	_, err = client.ListTagsForResource(ctx, &awskeyspaces.ListTagsForResourceInput{ResourceArn: ghost})
+	if !errors.As(err, &rnf) {
+		t.Fatalf("ListTagsForResource: got %v, want ResourceNotFoundException", err)
+	}
+
+	var ve *kstypes.ValidationException
+
+	_, err = client.TagResource(ctx, &awskeyspaces.TagResourceInput{
+		ResourceArn: aws.String("arn:aws:dynamodb:us-east-1:123456789012:table/x"), Tags: tags,
+	})
+	if !errors.As(err, &ve) {
+		t.Fatalf("TagResource non-cassandra ARN: got %v, want ValidationException", err)
+	}
+}
+
+// Tag restrictions over the wire: reserved aws: prefix, empty value, and more
+// than 60 tags in one request are ValidationException; the system_schema_mcs
+// keyspace exists and can't be deleted.
+func TestSDKTagRestrictionsAndSystemKeyspace(t *testing.T) {
+	client := newSDKClient(t)
+	ctx := context.Background()
+
+	out, err := client.CreateKeyspace(ctx, &awskeyspaces.CreateKeyspaceInput{KeyspaceName: aws.String("app")})
+	if err != nil {
+		t.Fatalf("CreateKeyspace: %v", err)
+	}
+
+	var ve *kstypes.ValidationException
+
+	for _, tag := range []kstypes.Tag{
+		{Key: aws.String("aws:owner"), Value: aws.String("v")},
+		{Key: aws.String("k"), Value: aws.String("")},
+	} {
+		_, err = client.TagResource(ctx, &awskeyspaces.TagResourceInput{ResourceArn: out.ResourceArn, Tags: []kstypes.Tag{tag}})
+		if !errors.As(err, &ve) {
+			t.Fatalf("TagResource %q: got %v, want ValidationException", aws.ToString(tag.Key), err)
+		}
+	}
+
+	many := make([]kstypes.Tag, 61)
+	for i := range many {
+		many[i] = kstypes.Tag{Key: aws.String("k" + strconv.Itoa(i)), Value: aws.String("v")}
+	}
+
+	_, err = client.TagResource(ctx, &awskeyspaces.TagResourceInput{ResourceArn: out.ResourceArn, Tags: many})
+	if !errors.As(err, &ve) {
+		t.Fatalf("TagResource 61 tags: got %v, want ValidationException", err)
+	}
+
+	_, err = client.CreateKeyspace(ctx, &awskeyspaces.CreateKeyspaceInput{KeyspaceName: aws.String("app2"), Tags: many})
+	if !errors.As(err, &ve) {
+		t.Fatalf("CreateKeyspace 61 tags: got %v, want ValidationException", err)
+	}
+
+	if _, err := client.GetKeyspace(ctx, &awskeyspaces.GetKeyspaceInput{KeyspaceName: aws.String("system_schema_mcs")}); err != nil {
+		t.Fatalf("GetKeyspace system_schema_mcs: %v", err)
+	}
+
+	_, err = client.DeleteKeyspace(ctx, &awskeyspaces.DeleteKeyspaceInput{KeyspaceName: aws.String("system_schema_mcs")})
+	if !errors.As(err, &ve) {
+		t.Fatalf("DeleteKeyspace system_schema_mcs: got %v, want ValidationException", err)
+	}
+}
diff --git a/server/aws/keyspaces/types.go b/server/aws/keyspaces/types.go
index 0dc7879e6..595ff1abb 100644
--- a/server/aws/keyspaces/types.go
+++ b/server/aws/keyspaces/types.go
@@ -30,12 +30,18 @@ func toWireSchema(s *ksdriver.SchemaDefinition) *types.SchemaDefinition {
 	return out
 }
 
+// toWireCapacity emits read/write capacity units only for PROVISIONED tables.
+// Real Keyspaces omits them for PAY_PER_REQUEST (their documented valid range
+// starts at 1, so 0 is never a real value).
 func toWireCapacity(c ksdriver.CapacitySpecification) *types.CapacitySpecificationSummary {
-	return &types.CapacitySpecificationSummary{
-		ThroughputMode:     types.ThroughputMode(c.ThroughputMode),
-		ReadCapacityUnits:  aws.Int64(c.ReadCapacityUnits),
-		WriteCapacityUnits: aws.Int64(c.WriteCapacityUnits),
+	out := &types.CapacitySpecificationSummary{ThroughputMode: types.ThroughputMode(c.ThroughputMode)}
+
+	if c.ThroughputMode == ksdriver.ThroughputProvisioned {
+		out.ReadCapacityUnits = aws.Int64(c.ReadCapacityUnits)
+		out.WriteCapacityUnits = aws.Int64(c.WriteCapacityUnits)
 	}
+
+	return out
 }
 
 func toWireEncryption(e ksdriver.EncryptionSpecification) *types.EncryptionSpecification {
diff --git a/server/aws/keyspaces/wirejson.go b/server/aws/keyspaces/wirejson.go
index 62312e986..118a56e61 100644
--- a/server/aws/keyspaces/wirejson.go
+++ b/server/aws/keyspaces/wirejson.go
@@ -32,23 +32,27 @@ func writeJSON(w http.ResponseWriter, v any) {
 	wire.WriteJSON(w, lowerCamelKeys(decoded))
 }
 
+// omitEmptyEnum reports whether an empty value should be dropped because real
+// Keyspaces omits the member when it is unset. It covers only
+// cdcSpecification.viewType: the SDK models it as a non-pointer string enum, so
+// encoding/json renders it as "" on a table without a CDC stream. In the
+// Keyspaces SDK, ViewType appears only on CdcSpecification and
+// CdcSpecificationSummary, so matching the key name is enough. This is not a
+// general rule for empty strings: the decoded map no longer carries Go types,
+// so a general rule would also drop non-enum members that are legitimately
+// empty. Add another unset enum here by name if one turns up.
+func omitEmptyEnum(key string, val any) bool {
+	return key == "ViewType" && val == ""
+}
+
 // lowerCamelKeys returns v with the first letter of every map key lowercased,
-// recursively. It also drops null values and the SDK's always-empty
-// resultMetadata envelope so the response carries only meaningful fields.
+// recursively. It also drops null values, an unset cdcSpecification.viewType
+// (see omitEmptyEnum), and the SDK's always-empty resultMetadata envelope so the
+// response carries only meaningful fields.
 func lowerCamelKeys(v any) any {
 	switch t := v.(type) {
 	case map[string]any:
-		out := make(map[string]any, len(t))
-
-		for k, val := range t {
-			if val == nil || k == "ResultMetadata" {
-				continue
-			}
-
-			out[lowerFirst(k)] = lowerCamelKeys(val)
-		}
-
-		return out
+		return lowerCamelMap(t)
 	case []any:
 		for i := range t {
 			t[i] = lowerCamelKeys(t[i])
@@ -69,6 +73,20 @@ func lowerCamelKeys(v any) any {
 	}
 }
 
+func lowerCamelMap(t map[string]any) map[string]any {
+	out := make(map[string]any, len(t))
+
+	for k, val := range t {
+		if val == nil || k == "ResultMetadata" || omitEmptyEnum(k, val) {
+			continue
+		}
+
+		out[lowerFirst(k)] = lowerCamelKeys(val)
+	}
+
+	return out
+}
+
 func lowerFirst(s string) string {
 	if s == "" {
 		return s