diff --git a/pkg/frontend/internal_executor.go b/pkg/frontend/internal_executor.go
index 9005467b5cd5b..4ea85f1bb97bb 100644
--- a/pkg/frontend/internal_executor.go
+++ b/pkg/frontend/internal_executor.go
@@ -211,7 +211,7 @@ func (ie *internalExecutor) ExecWithStatus(ctx context.Context, sql string, opts
 		ses:    sess,
 	}
 	defer tempExecCtx.Close()
-	err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql})
+	err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql, internalSQLSource: ie.usesDefaultSystemPrincipal(sess, opts)})
 	res := ie.proto.swapOutResult()
 	status.AffectedRows = res.affectedRows
 	if err != nil {
@@ -240,7 +240,7 @@ func (ie *internalExecutor) Query(ctx context.Context, sql string, opts ie.Sessi
 		ses:    sess,
 	}
 	defer tempExecCtx.Close()
-	err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql})
+	err = doComQuery(sess, &tempExecCtx, &UserInput{sql: sql, internalSQLSource: ie.usesDefaultSystemPrincipal(sess, opts)})
 	res := ie.proto.swapOutResult()
 	res.err = moerr.AttachCause(ctx, err)
 	return res
@@ -294,7 +294,20 @@ func (ie *internalExecutor) newCmdSession(ctx context.Context, opts ie.SessionOv
 	now, _ := runtime.ServiceRuntime(ie.service).Clock().Now()
 	sess.lastCommitTS = now
 
-	if ie.baseSessOpts.Username != nil || opts.Username != nil {
+	if ie.usesDefaultSystemPrincipal(sess, opts) {
+		// The unnamed system administrator uses reserved user ID 0, whose catalog
+		// name is root, not the synthetic SQL-source name "internal". Bind
+		// after both override layers without changing explicit definers or
+		// bypassing current-principal/role authorization. Other selected roles
+		// keep their existing fail-closed identity rather than borrowing moadmin.
+		t.mu.Lock()
+		t.Tenant = sysAccountName
+		t.User = rootName
+		if t.DefaultRoleID == moAdminRoleID {
+			t.DefaultRole = moAdminRoleName
+		}
+		t.mu.Unlock()
+	} else if ie.baseSessOpts.Username != nil || opts.Username != nil {
 		if err := bindInternalSessionPrincipal(ctx, sess); err != nil {
 			sess.Close()
 			return nil, err
@@ -305,6 +318,13 @@ func (ie *internalExecutor) newCmdSession(ctx context.Context, opts ie.SessionOv
 	return sess, nil
 }
 
+func (ie *internalExecutor) usesDefaultSystemPrincipal(sess *Session, opts ie.SessionOverrideOptions) bool {
+	principal := sess.GetTenantInfo()
+	return ie.baseSessOpts.Username == nil && opts.Username == nil &&
+		principal.GetTenantID() == sysAccountID && principal.GetUserID() == rootID &&
+		principal.GetDefaultRoleID() == moAdminRoleID
+}
+
 func (ie *internalExecutor) ApplySessionOverride(opts ie.SessionOverrideOptions) {
 	ie.baseSessOpts = opts
 }
diff --git a/pkg/frontend/internal_executor_test.go b/pkg/frontend/internal_executor_test.go
index 62907355801fb..f39333569cb2b 100644
--- a/pkg/frontend/internal_executor_test.go
+++ b/pkg/frontend/internal_executor_test.go
@@ -16,11 +16,14 @@ package frontend
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/prashantv/gostub"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/tidwall/btree"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
@@ -29,6 +32,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/defines"
+	"github.com/matrixorigin/matrixone/pkg/frontend/constant"
 	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/testutil"
 	ie "github.com/matrixorigin/matrixone/pkg/util/internalExecutor"
@@ -87,6 +91,143 @@ func TestInternalExecutorAuthorizationIdentity(t *testing.T) {
 		bh.sql2result[getSqlForRoleNameOfRoleId(99)] = newMrsForRestoreStringRows([]string{"role_name"}, nil)
 		stub := gostub.StubFunc(&NewBackgroundExec, bh)
 		defer stub.Reset()
+		checkSource := func(t *testing.T, sess *Session, defaultSystem, wantInternal bool) {
+			t.Helper()
+			input := &UserInput{sql: "/* cloud_nonuser */ select 1; select 2", internalSQLSource: defaultSystem}
+			require.False(t, input.isInternal(), "source provenance must not become an internal AST input")
+			require.False(t, sess.GetIsInternal(), "source provenance must not bypass authorization")
+			require.Equal(t, wantInternal, input.isInternalSQLSource(sess))
+			fragments, sources, err := schedulingSQLByStatementWithSQLMode(t.Context(), input.sql, "", input.isInternalSQLSource(sess))
+			require.NoError(t, err)
+			require.Len(t, fragments, 2)
+			wantSources := []string{constant.CloudNoUserSql, constant.ExternSql}
+			plainSource := constant.ExternSql
+			if wantInternal {
+				wantSources = []string{constant.InternalSql, constant.InternalSql}
+				plainSource = constant.InternalSql
+			}
+			require.Equal(t, wantSources, sources)
+			input.setSqlSourceTypes(sess, sources)
+			require.Equal(t, wantSources, input.getSqlSourceTypes())
+			plain := &UserInput{sql: "select 1", internalSQLSource: defaultSystem}
+			plain.genSqlSourceType(sess)
+			require.Equal(t, []string{plainSource}, plain.getSqlSourceTypes())
+		}
+		// The default numeric sys/root/moadmin identity must agree with the username
+		// checked by authorization. This must not turn internal execution into
+		// an authorization bypass or replace any explicitly supplied principal.
+		sysContext := defines.AttachAccount(t.Context(), 0, 0, 0)
+		tenantContext := defines.AttachAccount(t.Context(), 10, 42, 43)
+		for _, tc := range []struct {
+			name                      string
+			ctx                       context.Context
+			base, command             ie.SessionOverrideOptions
+			user                      string
+			accountID, userID, roleID uint32
+			defaultSystem             bool
+		}{
+			{name: "default without context identity", ctx: t.Context(), user: "root", defaultSystem: true},
+			{name: "default sys account only", ctx: defines.AttachAccountId(t.Context(), 0), user: "root", defaultSystem: true},
+			{name: "default sys zero IDs", ctx: sysContext, user: "root", defaultSystem: true},
+			{name: "default sys public role stays fail-closed", ctx: defines.AttachAccount(t.Context(), 0, 0, publicRoleID), user: "internal", roleID: publicRoleID},
+			{name: "default sys selected role stays fail-closed", ctx: defines.AttachAccount(t.Context(), 0, 0, 43), user: "internal", roleID: 43},
+			{name: "non-sys zero user is not root", ctx: defines.AttachAccount(t.Context(), 10, 0, 43), accountID: 10, roleID: 43},
+			{name: "sys nonzero user is not root", ctx: defines.AttachAccount(t.Context(), 0, 42, 43), user: "internal", userID: 42, roleID: 43},
+			{
+				name: "base changes account", ctx: sysContext,
+				base: ie.NewOptsBuilder().AccountId(10).DefaultRoleId(43).Finish(),
+				user: "internal", accountID: 10, roleID: 43,
+			},
+			{
+				name: "command changes account", ctx: sysContext,
+				command: ie.NewOptsBuilder().AccountId(10).DefaultRoleId(43).Finish(),
+				user:    "internal", accountID: 10, roleID: 43,
+			},
+			{
+				name: "base changes user", ctx: sysContext,
+				base: ie.NewOptsBuilder().UserId(42).DefaultRoleId(43).Finish(),
+				user: "internal", userID: 42, roleID: 43,
+			},
+			{
+				name: "command changes user", ctx: sysContext,
+				command: ie.NewOptsBuilder().UserId(42).DefaultRoleId(43).Finish(),
+				user:    "internal", userID: 42, roleID: 43,
+			},
+			{
+				name: "base restores sys root but preserves non-admin role", ctx: tenantContext,
+				base:   ie.NewOptsBuilder().AccountId(0).UserId(0).Finish(),
+				roleID: 43,
+			},
+			{
+				name: "command restores sys root but preserves non-admin role", ctx: tenantContext,
+				command: ie.NewOptsBuilder().AccountId(0).UserId(0).Finish(),
+				roleID:  43,
+			},
+			{
+				name: "base restores full sys root moadmin identity", ctx: tenantContext,
+				base: ie.NewOptsBuilder().AccountId(0).UserId(0).DefaultRoleId(0).Finish(),
+				user: "root", defaultSystem: true,
+			},
+			{
+				name: "explicit base internal stays internal", ctx: sysContext,
+				base: ie.NewOptsBuilder().Username("internal").DefaultRoleId(43).Finish(),
+				user: "internal", roleID: 43,
+			},
+			{
+				name: "explicit command internal stays internal", ctx: sysContext,
+				command: ie.NewOptsBuilder().Username("sys:internal:moadmin").DefaultRoleId(43).Finish(),
+				user:    "internal", roleID: 43,
+			},
+			{
+				name: "command restores root after base numeric identity", ctx: sysContext,
+				base:    ie.NewOptsBuilder().AccountId(10).UserId(42).DefaultRoleId(43).Finish(),
+				command: ie.NewOptsBuilder().AccountId(0).UserId(0).DefaultRoleId(0).Finish(),
+				user:    "root", defaultSystem: true,
+			},
+			{
+				name: "command overrides base root IDs", ctx: sysContext,
+				base:    ie.NewOptsBuilder().AccountId(0).UserId(0).DefaultRoleId(43).Finish(),
+				command: ie.NewOptsBuilder().AccountId(10).UserId(42).Finish(),
+				user:    "internal", accountID: 10, userID: 42, roleID: 43,
+			},
+			{
+				name: "numeric command retains explicit base name", ctx: tenantContext,
+				base:    ie.NewOptsBuilder().Username("internal").Finish(),
+				command: ie.NewOptsBuilder().AccountId(0).UserId(0).Finish(),
+				user:    "internal", roleID: 43,
+			},
+			{
+				name: "command explicit name overrides base root", ctx: sysContext,
+				base:    ie.NewOptsBuilder().Username("root").Finish(),
+				command: ie.NewOptsBuilder().Username("internal").DefaultRoleId(43).Finish(),
+				user:    "internal", roleID: 43,
+			},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				executor := newIe("")
+				executor.ApplySessionOverride(tc.base)
+				sess, err := executor.newCmdSession(tc.ctx, tc.command)
+				require.NoError(t, err)
+				defer sess.Close()
+				principal := sess.GetTenantInfo()
+				require.Equal(t, tc.user, principal.GetUser())
+				require.Equal(t, tc.accountID, principal.GetTenantID())
+				require.Equal(t, tc.userID, principal.GetUserID())
+				require.Equal(t, tc.roleID, principal.GetDefaultRoleID())
+				require.False(t, sess.GetIsInternal(), "normalization must not bypass catalog authorization")
+				defaultSystem := executor.usesDefaultSystemPrincipal(sess, tc.command)
+				require.Equal(t, tc.defaultSystem, defaultSystem)
+				if tc.defaultSystem {
+					require.Equal(t, sysAccountName, principal.GetTenant())
+				}
+				checkSource(t, sess, defaultSystem, tc.defaultSystem || tc.user == "internal")
+				for _, trusted := range []bool{false, true} {
+					authSQL := getSqlForActiveRolesForAuthorization(principal, trusted)
+					require.Contains(t, authSQL, fmt.Sprintf("u.user_id = %d", tc.userID))
+					require.Contains(t, authSQL, "u.user_name = '"+tc.user+"'")
+				}
+			})
+		}
 		for _, username := range []string{"definer", "tenant:definer", "tenant:definer:writer", "tenant#definer#writer", "tenant%3Adefiner%3Awriter", "tenant:definer:writer?label=value"} {
 			t.Run(username, func(t *testing.T) {
 				executor := newIe("")
@@ -105,6 +246,8 @@ func TestInternalExecutorAuthorizationIdentity(t *testing.T) {
 				require.Equal(t, "tenant", principal.GetTenant())
 				require.Equal(t, "selected_writer", principal.GetDefaultRole(), "selected numeric role, not login role")
 				require.Contains(t, getSqlForActiveRolesForAuthorization(principal, false), "u.user_name = 'definer'")
+				require.False(t, executor.usesDefaultSystemPrincipal(sess, opts))
+				checkSource(t, sess, false, false)
 			})
 		}
 		t.Run("root zero IDs", func(t *testing.T) {
@@ -118,7 +261,76 @@ func TestInternalExecutorAuthorizationIdentity(t *testing.T) {
 			require.Zero(t, sess.GetTenantInfo().GetTenantID())
 			require.Zero(t, sess.GetTenantInfo().GetUserID())
 			require.Zero(t, sess.GetTenantInfo().GetDefaultRoleID())
+			require.False(t, executor.usesDefaultSystemPrincipal(sess, opts))
+			checkSource(t, sess, false, false)
 		})
+		for _, tc := range []struct {
+			name, externalUser string
+			roleID             uint32
+			present            bool
+			failure            error
+		}{
+			{name: "default root catalog present", present: true},
+			{name: "default root catalog missing"},
+			{name: "default root catalog lookup error", failure: errors.New("catalog lookup failed")},
+			{name: "default public role cannot normalize into root", roleID: publicRoleID},
+			{name: "default selected role cannot normalize into root", roleID: 43},
+			{name: "external root name is not an authorization bypass", externalUser: "root"},
+			{name: "external internal name is not an authorization bypass", externalUser: "internal"},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				var sess *Session
+				if tc.externalUser == "" {
+					executor := newIe("")
+					opts := ie.NewOptsBuilder().Finish()
+					var err error
+					sess, err = executor.newCmdSession(defines.AttachAccount(t.Context(), sysAccountID, rootID, tc.roleID), opts)
+					require.NoError(t, err)
+					defer sess.Close()
+					defaultSystem := executor.usesDefaultSystemPrincipal(sess, opts)
+					require.Equal(t, tc.roleID == moAdminRoleID, defaultSystem)
+					checkSource(t, sess, defaultSystem, true)
+				} else {
+					// An ordinary session has neither the executor's origin nor
+					// isInternal set, even when its login name is "internal".
+					sess = &Session{}
+					sess.SetTenantInfo(&TenantInfo{Tenant: "sys", User: tc.externalUser, DefaultRole: moAdminRoleName})
+					checkSource(t, sess, false, tc.externalUser == "internal")
+				}
+				authBH := &backgroundExecTest{}
+				authBH.init()
+				require.Equal(t, tc.roleID, sess.GetTenantInfo().GetDefaultRoleID())
+				query := getSqlForActiveRolesForAuthorization(sess.GetTenantInfo(), tc.roleID == moAdminRoleID || tc.roleID == publicRoleID)
+				wantUser := tc.externalUser
+				if wantUser == "" {
+					wantUser = "root"
+					if tc.roleID != moAdminRoleID {
+						wantUser = "internal"
+					}
+				}
+				require.Contains(t, query, "u.user_id = 0 and u.user_name = '"+wantUser+"'")
+				var rows [][]interface{}
+				if tc.present {
+					rows = [][]interface{}{{int64(0)}}
+				}
+				authBH.sql2result[query] = newMrsForRoleIdOfUserId(rows)
+				if tc.failure != nil {
+					authBH.sql2err[query] = tc.failure
+				}
+				roles := &btree.Set[int64]{}
+				err := loadActiveRolesForAuthorization(t.Context(), authBH, sess, roles)
+				switch {
+				case tc.failure != nil:
+					require.ErrorIs(t, err, tc.failure)
+				case !tc.present:
+					require.ErrorContains(t, err, "authenticated user no longer matches current catalog")
+				default:
+					require.NoError(t, err)
+				}
+				require.Equal(t, tc.present, roles.Contains(int64(tc.roleID)))
+				require.Equal(t, []string{query}, authBH.executedSQLs, "default root must still validate current catalog identity")
+			})
+		}
 		for _, username := range []string{"tenant:", "tenant::writer"} {
 			for _, base := range []bool{false, true} {
 				t.Run("invalid/"+username+map[bool]string{false: "/command", true: "/base"}[base], func(t *testing.T) {
diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go
index 9b4e453c8604d..09f5dcfe6afc7 100644
--- a/pkg/frontend/mysql_cmd_executor.go
+++ b/pkg/frontend/mysql_cmd_executor.go
@@ -1396,6 +1396,9 @@ func executeAnalyzeDerivedQuery(ses *Session, outerExecCtx *ExecCtx, sql string)
 		rewritePolicy:             policy,
 		rewritePolicyMaterialized: true,
 	}
+	if outerExecCtx.input != nil {
+		derivedInput.internalSQLSource = outerExecCtx.input.internalSQLSource
+	}
 	if err := doComQuery(ses, &tempExecCtx, derivedInput); err != nil {
 		return nil, err
 	}
diff --git a/pkg/frontend/mysql_cmd_executor_test.go b/pkg/frontend/mysql_cmd_executor_test.go
index 3bb2b18cd83b1..37de34723e1d7 100644
--- a/pkg/frontend/mysql_cmd_executor_test.go
+++ b/pkg/frontend/mysql_cmd_executor_test.go
@@ -2614,6 +2614,29 @@ func TestSQLModeStagingDefersRewriteWithRequestSnapshot(t *testing.T) {
 	bindSource(third)
 	require.Equal(t, constant.CloudNoUserSql, third.getSqlSourceType(0))
 	require.Nil(t, input.sqlSourceType, "the unexecuted remainder must not be classified under the old mode")
+	for _, internalSource := range []bool{false, true} {
+		t.Run(fmt.Sprintf("source provenance %v", internalSource), func(t *testing.T) {
+			request := &UserInput{sql: "select * from src.t", rewritePolicy: policy, internalSQLSource: internalSource}
+			statement := newSQLStatementInput(request, request.sql)
+			require.NotSame(t, request, statement)
+			rewritten, err := rewriteSQLStatementInput(ctx, ses, statement)
+			require.NoError(t, err)
+			require.NotSame(t, statement, rewritten)
+			assertMaterializedRemap(t, ctx, rewritten.sql, map[string]string{"src": "dst"})
+			want := constant.ExternSql
+			if internalSource {
+				want = constant.InternalSql
+			}
+			for _, copied := range []*UserInput{statement, rewritten} {
+				require.Equal(t, internalSource, copied.internalSQLSource)
+				require.False(t, copied.isInternal(), "source propagation must not enable internal-input authorization")
+				copied.genSqlSourceType(ses)
+				require.Equal(t, []string{want}, copied.getSqlSourceTypes())
+			}
+			require.False(t, ses.GetIsInternal())
+			require.Nil(t, request.sqlSourceType, "derived source classification must not mutate the request")
+		})
+	}
 }
 
 func assertMaterializedRemap(t *testing.T, ctx context.Context, sql string, want map[string]string) {
@@ -3459,11 +3482,12 @@ func TestExecuteAnalyzeDerivedQueryRestoresResponderOnSuccess(t *testing.T) {
 	defer ctrl.Finish()
 
 	ses := newTestSession(t, ctrl)
+	defer ses.Close()
 	writer := &countingMysqlWriter{testMysqlWriter: &testMysqlWriter{}}
 	live := NewMysqlResp(writer)
 	ses.ReplaceResponser(live)
 	ctx := defines.AttachAccountId(context.Background(), sysAccountID)
-	outerExecCtx := &ExecCtx{reqCtx: ctx, ses: ses}
+	outerExecCtx := &ExecCtx{reqCtx: ctx, ses: ses, input: &UserInput{internalSQLSource: true}}
 	eng := mock_frontend.NewMockEngine(ctrl)
 	eng.EXPECT().Hints().Return(engine.Hints{CommitOrRollbackTimeout: time.Second}).AnyTimes()
 	txnOperator := mock_frontend.NewMockTxnOperator(ctrl)
@@ -3471,7 +3495,17 @@ func TestExecuteAnalyzeDerivedQueryRestoresResponderOnSuccess(t *testing.T) {
 	txnOperator.EXPECT().GetWorkspace().Return(newTestWorkspace()).AnyTimes()
 	txnOperator.EXPECT().SetFootPrints(gomock.Any(), gomock.Any()).AnyTimes()
 	txnOperator.EXPECT().Status().Return(txn.TxnStatus_Active).AnyTimes()
-	txnOperator.EXPECT().TryEnterRunSqlWithTokenAndSQL(gomock.Any(), gomock.Any()).Return(uint64(0), nil).AnyTimes()
+	observedDerivedSource := false
+	txnOperator.EXPECT().TryEnterRunSqlWithTokenAndSQL(gomock.Any(), gomock.Any()).DoAndReturn(func(context.CancelFunc, string) (uint64, error) {
+		derived := ses.GetTxnCompileCtx().execCtx.input
+		require.NotNil(t, derived)
+		require.True(t, derived.internalSQLSource)
+		require.False(t, derived.isInternal())
+		require.False(t, ses.GetIsInternal())
+		require.Equal(t, constant.InternalSql, derived.getSqlSourceType(0))
+		observedDerivedSource = true
+		return 0, nil
+	}).AnyTimes()
 	txnOperator.EXPECT().ExitRunSqlWithToken(gomock.Any()).AnyTimes()
 	txnOperator.EXPECT().NextSequence().Return(uint64(0)).AnyTimes()
 	txnOperator.EXPECT().Commit(gomock.Any()).Return(nil).AnyTimes()
@@ -3482,6 +3516,7 @@ func TestExecuteAnalyzeDerivedQueryRestoresResponderOnSuccess(t *testing.T) {
 
 	result, err := executeAnalyzeDerivedQuery(ses, outerExecCtx, "select 1")
 	require.NoError(t, err)
+	require.True(t, observedDerivedSource, "derived execution must exercise the provenance assertion")
 	require.NotNil(t, result)
 	require.Equal(t, uint64(1), result.GetColumnCount())
 	require.Equal(t, uint64(1), result.GetRowCount())
diff --git a/pkg/frontend/util.go b/pkg/frontend/util.go
index 8822b0cd8d24c..51e4a06ed4836 100644
--- a/pkg/frontend/util.go
+++ b/pkg/frontend/util.go
@@ -1835,6 +1835,9 @@ type UserInput struct {
 	// replace old logic: (stmt != nil)
 	// cc isInternal()
 	isInternalInput bool
+	// internalSQLSource preserves unnamed system-executor provenance without
+	// changing isInternalInput's parser, diagnostics or query-ID semantics.
+	internalSQLSource bool
 	// operator account, the account executes restoration
 	// e.g. sys takes a snapshot sn1 for acc1, then restores acc1 from snapshot sn1. In this scenario, sys is the operator account
 	isRestoreByTs bool
@@ -1899,7 +1902,7 @@ func (ui *UserInput) genSqlSourceType(ses FeSession) {
 
 func (ui *UserInput) isInternalSQLSource(ses FeSession) bool {
 	tenant := ses.GetTenantInfo()
-	internal := ui.isInternal() || tenant == nil || strings.HasPrefix(ui.getSql(), cmdFieldListSql)
+	internal := ui.internalSQLSource || ui.isInternal() || tenant == nil || strings.HasPrefix(ui.getSql(), cmdFieldListSql)
 	if tenant != nil {
 		special, _, _ := isSpecialUser(tenant.GetUser())
 		internal = internal || special || tenant.GetTenant() == sysAccountName && tenant.GetUser() == "internal"
diff --git a/pkg/tests/issues/internal_executor_system_identity_test.go b/pkg/tests/issues/internal_executor_system_identity_test.go
new file mode 100644
index 0000000000000..6005575c5a290
--- /dev/null
+++ b/pkg/tests/issues/internal_executor_system_identity_test.go
@@ -0,0 +1,93 @@
+// Copyright 2026 Matrix Origin
+//
+// 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
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package issues
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/defines"
+	"github.com/matrixorigin/matrixone/pkg/embed"
+	"github.com/matrixorigin/matrixone/pkg/frontend"
+	ie "github.com/matrixorigin/matrixone/pkg/util/internalExecutor"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInternalExecutorSystemIdentity(t *testing.T) {
+	runAuthenticatedClusterTest(t, func(c embed.Cluster) {
+		ctx, cancel := context.WithTimeout(t.Context(), 120*time.Second)
+		defer cancel()
+		cn, err := c.GetCNService(0)
+		require.NoError(t, err)
+		sys, err := sql.Open("mysql", fmt.Sprintf("dump:111@tcp(127.0.0.1:%d)/", cn.GetServiceConfig().CN.Frontend.Port))
+		require.NoError(t, err)
+		defer func() { require.NoError(t, sys.Close()) }()
+		const database = "internal_executor_system_identity"
+		defer func() {
+			cleanup, done := context.WithTimeout(context.Background(), 30*time.Second)
+			defer done()
+			execSQLRequire(t, cleanup, sys, "drop database if exists "+database)
+		}()
+		execSQLRequire(t, ctx, sys, "create database "+database)
+		execSQLRequire(t, ctx, sys, "create table "+database+".t(id int primary key)")
+		internal := frontend.NewInternalExecutor(cn.ServiceID())
+		for i, tc := range []struct {
+			name string
+			ctx  context.Context
+		}{
+			{"sys account", defines.AttachAccountId(ctx, 0)},
+			{"explicit system IDs", defines.AttachAccount(ctx, 0, 0, 0)},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				result := internal.Query(tc.ctx, "select user_name from mo_catalog.mo_user where user_id=0", ie.SessionOverrideOptions{})
+				require.NoError(t, result.Error())
+				require.Equal(t, uint64(1), result.RowCount())
+				user, err := result.GetString(tc.ctx, 0, 0)
+				require.NoError(t, err)
+				require.Equal(t, "root", user)
+				require.NoError(t, internal.Exec(tc.ctx, fmt.Sprintf("insert into %s.t values(%d)", database, i+1), ie.SessionOverrideOptions{}))
+				result = internal.Query(tc.ctx, fmt.Sprintf("select id from %s.t where id=%d", database, i+1), ie.SessionOverrideOptions{})
+				require.NoError(t, result.Error())
+				require.Equal(t, uint64(1), result.RowCount())
+				id, err := result.GetUint64(tc.ctx, 0, 0)
+				require.NoError(t, err)
+				require.Equal(t, uint64(i+1), id)
+			})
+		}
+		// A caller-supplied name is not the anonymous system identity, even
+		// when it carries root's numeric IDs. Retain the real catalog check.
+		t.Run("explicit mismatched name remains denied", func(t *testing.T) {
+			sysCtx := defines.AttachAccount(ctx, 0, 0, 0)
+			opts := ie.NewOptsBuilder().Username("internal").AccountId(0).UserId(0).DefaultRoleId(0).Finish()
+			result := internal.Query(sysCtx, "select id from "+database+".t", opts)
+			require.ErrorContains(t, result.Error(), "authenticated user no longer matches current catalog")
+			require.ErrorContains(t, internal.Exec(sysCtx, "insert into "+database+".t values(99)", opts), "authenticated user no longer matches current catalog")
+			result = internal.Query(sysCtx, "select count(*) from "+database+".t", ie.SessionOverrideOptions{})
+			require.NoError(t, result.Error(), "default identity must recover after an explicit identity fails")
+			count, err := result.GetUint64(sysCtx, 0, 0)
+			require.NoError(t, err)
+			require.Equal(t, uint64(2), count, "denied writes must have no side effects")
+		})
+		t.Run("non-administrator role cannot borrow default administrator identity", func(t *testing.T) {
+			publicCtx := defines.AttachAccount(ctx, 0, 0, 1)
+			result := internal.Query(publicCtx, "show accounts", ie.SessionOverrideOptions{})
+			require.ErrorContains(t, result.Error(), "do not have privilege")
+			require.ErrorContains(t, internal.Exec(publicCtx, "insert into "+database+".t values(99)", ie.SessionOverrideOptions{}), "do not have privilege")
+		})
+	})
+}