diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala
index aa60f11c8e4..237eb1cd537 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceSearchQueryBuilder.scala
@@ -84,25 +84,24 @@ abstract class VersionedResourceSearchQueryBuilder[Rec <: Record, P](
 
   /**
     * `uid` is null for anonymous callers. Visibility: public only when `uid` is null;
-    * explicitly-granted only when `includePublic` is false; both when it is true.
+    * owned or explicitly granted only when `includePublic` is false; both when it is true.
     */
   override protected def constructFromClause(
       uid: Integer,
       params: DashboardResource.SearchQueryParams,
       includePublic: Boolean = false
   ): TableLike[_] = {
-    val baseJoin = tables.joinWithAccessAndOwner(
-      Some(if (uid == null) DSL.falseCondition() else tables.access.uidColumn.eq(uid))
-    )
+    val baseJoin = tables.joinWithViewerAccessAndOwner(uid)
 
     val condition: Condition =
       if (uid == null) {
         tables.isPublicColumn.eq(true)
       } else {
+        val ownedOrGranted = tables.access.uidColumn.isNotNull.or(tables.ownerUidColumn.eq(uid))
         if (includePublic) {
-          tables.isPublicColumn.eq(true).or(tables.access.uidColumn.isNotNull)
+          tables.isPublicColumn.eq(true).or(ownedOrGranted)
         } else {
-          tables.access.uidColumn.isNotNull
+          ownedOrGranted
         }
       }
     baseJoin.where(condition)
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTables.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTables.scala
index 7a6d4232156..1038ababc9a 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTables.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTables.scala
@@ -22,6 +22,7 @@ package org.apache.texera.web.resource.dashboard
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
 import org.apache.texera.dao.jooq.generated.Tables.{DATASET, MODEL}
+import org.apache.texera.auth.util.OwnerAccess
 import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
 import org.apache.texera.dao.jooq.generated.tables.User.USER
 import org.apache.texera.dao.jooq.generated.tables.pojos.{Dataset, Model, User}
@@ -34,6 +35,7 @@ import org.apache.texera.web.resource.dashboard.hub.EntityTables.{AccessTable, B
 import org.apache.texera.web.resource.dashboard.user.dataset.DatasetResource.DashboardDataset
 import org.apache.texera.web.resource.dashboard.user.model.ModelResource.DashboardModel
 import org.jooq._
+import org.jooq.impl.DSL
 
 import java.sql.Timestamp
 
@@ -84,19 +86,25 @@ trait VersionedResourceTables[Rec <: Record, P] extends BaseEntityTable with Laz
       .on(USER.UID.eq(ownerUidColumn))
   }
 
+  /** The access join narrowed to `uid`'s own grant, or to none for an anonymous viewer. */
+  final def joinWithViewerAccessAndOwner(uid: Integer): Table[Record] =
+    joinWithAccessAndOwner(
+      Some(if (uid == null) DSL.falseCondition() else access.uidColumn.eq(uid))
+    )
+
   /** `None` when LakeFS cannot size the resource; callers drop it and report a mismatch. */
   final def hydrate(
       record: Record,
       uid: Integer
   ): Option[(Integer, DashboardClickableFileEntry)] = {
     val resource = pojo(record)
+    val isOwner = ownerUidOf(resource) == uid
     repositorySize(resource).map { size =>
       idOf(resource) -> entry(
         resource,
         record.into(USER).into(classOf[User]).getEmail,
-        Option(record.get(access.privilegeColumn, classOf[PrivilegeEnum]))
-          .getOrElse(PrivilegeEnum.NONE),
-        ownerUidOf(resource) == uid,
+        OwnerAccess.privilege(isOwner, record.get(access.privilegeColumn, classOf[PrivilegeEnum])),
+        isOwner,
         size
       )
     }
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala
index 42ae99d9ce3..973f41882ab 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/hub/HubResource.scala
@@ -276,10 +276,18 @@ object HubResource {
     mapWorkflowEntries(records, uid)
   }
 
-  /**
-    * Hydrates ids of one LakeFS-backed resource into hub entries. Unsizable resources are
-    * dropped; ids are de-duplicated because the access join can match one twice.
-    */
+  /** The access join is narrowed to `uid`, so each id yields one row carrying only the viewer's grant. */
+  private[hub] def versionedResourcesByIdsQuery(
+      tables: VersionedResourceTables[_ <: Record, _],
+      ids: Seq[Integer],
+      uid: Integer
+  ) =
+    context
+      .select()
+      .from(tables.joinWithViewerAccessAndOwner(uid))
+      .where(tables.idColumn.in(ids: _*))
+
+  /** Hydrates ids of one LakeFS-backed resource into hub entries; unsizable resources are dropped. */
   def fetchDashboardVersionedResourcesByIds(
       tables: VersionedResourceTables[_ <: Record, _],
       ids: Seq[Integer],
@@ -289,27 +297,12 @@ object HubResource {
       return List.empty[DashboardClickableFileEntry]
     }
 
-    val records = context
-      .select()
-      .from(tables.joinWithAccessAndOwner(None))
-      .where(tables.idColumn.in(ids: _*))
-      .groupBy(
-        tables.idColumn,
-        tables.nameColumn,
-        tables.descriptionColumn,
-        tables.ownerUidColumn,
-        USER.NAME,
-        tables.access.idColumn,
-        tables.access.uidColumn,
-        USER.UID
-      )
+    versionedResourcesByIdsQuery(tables, ids, uid)
       .fetch()
-
-    records.asScala
+      .asScala
       .flatMap(record => tables.hydrate(record, uid))
-      .toList
-      .distinctBy(_._1)
       .map(_._2)
+      .toList
   }
 }
 
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala
index 8f975f90413..f9971e54924 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala
@@ -21,6 +21,7 @@ package org.apache.texera.web.resource.dashboard.user.workflow
 
 import io.dropwizard.auth.Auth
 import org.apache.texera.auth.SessionUser
+import org.apache.texera.auth.util.OwnerAccess
 import org.apache.texera.dao.SqlServer
 import org.apache.texera.dao.jooq.generated.Tables._
 import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
@@ -199,7 +200,8 @@ class WorkflowAccessResource() {
       .where(WORKFLOW_OF_USER.WID.eq(wid))
       .fetchOneInto(classOf[Integer])
     if (userUid == workflowOwnerUid) {
-      throw new ForbiddenException("You cannot modify the owner's permissions!")
+      // amber is on javax JAX-RS, so it raises its own ForbiddenException with the shared message.
+      throw new ForbiddenException(OwnerAccess.ownerChangeRefused)
     }
 
     try {
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/DatasetSearchQueryBuilderSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/DatasetSearchQueryBuilderSpec.scala
index 51028942c0f..ede4a858722 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/DatasetSearchQueryBuilderSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/DatasetSearchQueryBuilderSpec.scala
@@ -324,6 +324,18 @@ class DatasetSearchQueryBuilderSpec
     } finally shutdownDB()
   }
 
+  /** Runs `body` with the owner's own grant row replaced by `privilege`, or removed, then restores it. */
+  private def withOwnerRow(privilege: Option[PrivilegeEnum])(body: => Unit): Unit = {
+    val dao = new DatasetUserAccessDao(getDSLContext.configuration())
+    dao.delete(grant(sizedDid, ownerUid, PrivilegeEnum.WRITE))
+    privilege.foreach(p => dao.insert(grant(sizedDid, ownerUid, p)))
+    try body
+    finally {
+      privilege.foreach(p => dao.delete(grant(sizedDid, ownerUid, p)))
+      dao.insert(grant(sizedDid, ownerUid, PrivilegeEnum.WRITE))
+    }
+  }
+
   private def params(keywords: String*): SearchQueryParams =
     SearchQueryParams(keywords = keywords.toList.asJava)
 
@@ -391,6 +403,16 @@ class DatasetSearchQueryBuilderSpec
     sql should not include "is_public = true"
   }
 
+  it should "still return the caller's own dataset when they have no grant row" in {
+    withOwnerRow(None) {
+      val p = SearchQueryParams(datasetIds = List(sizedDid).asJava)
+      DatasetSearchQueryBuilder
+        .constructQuery(ownerUid, p, includePublic = false)
+        .fetch()
+        .size() shouldBe 1
+    }
+  }
+
   it should "add public datasets to the granted ones when they are included" in {
     val sql = sqlFor(uid, includePublic = true)
 
@@ -651,6 +673,14 @@ class DatasetSearchQueryBuilderSpec
     entryFor(otherUid, sizedDid).dataset.value.accessPrivilege shouldBe PrivilegeEnum.NONE
   }
 
+  it should "report WRITE for the owner even when their own grant row says READ" in {
+    requireStub()
+
+    withOwnerRow(Some(PrivilegeEnum.READ)) {
+      entryFor(ownerUid, sizedDid).dataset.value.accessPrivilege shouldBe PrivilegeEnum.WRITE
+    }
+  }
+
   it should "tag the entry as a dataset and fill the dataset payload slot" in {
     // `entry.workflow shouldBe None` used to sit here and was vacuous:
     // `DashboardClickableFileEntry` declares it `= None` and this file does not pass it, so no
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTablesSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTablesSpec.scala
index 8c990576e11..2d2dad0136a 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTablesSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/VersionedResourceTablesSpec.scala
@@ -146,6 +146,19 @@ class VersionedResourceTablesSpec
     sql should include("dataset_user_access.uid = 42")
   }
 
+  "joinWithViewerAccessAndOwner" should "narrow the access rows to the viewer, or to none when anonymous" in {
+    val forViewer =
+      rendered(
+        getDSLContext.renderInlined(tables.joinWithViewerAccessAndOwner(Integer.valueOf(42)))
+      )
+    forViewer should include("dataset_user_access.uid = 42")
+
+    val forAnonymous =
+      rendered(getDSLContext.renderInlined(tables.joinWithViewerAccessAndOwner(null)))
+    forAnonymous should not include "dataset_user_access.uid ="
+    forAnonymous should include("false")
+  }
+
   "ModelTables" should "name every column the shared query logic reads" in {
     modelTables.table.getName shouldBe "model"
     modelTables.idColumn.getName shouldBe "mid"
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala
index 695cf9bf469..ea23a769ae6 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala
@@ -949,6 +949,31 @@ class HubResourceSpec
     byEntity shouldBe Map((Wf, 810805) -> 1, (Ds, 820802) -> 2)
   }
 
+  "versionedResourcesByIdsQuery" should "carry only the viewer's own grant, never another user's" in {
+    val did = seedDataset(820860, "ds_two_grants")
+    for ((uid, privilege) <- Seq(ownerUid -> PrivilegeEnum.WRITE, likerUid -> PrivilegeEnum.READ)) {
+      getDSLContext
+        .insertInto(DATASET_USER_ACCESS)
+        .set(DATASET_USER_ACCESS.DID, did)
+        .set(DATASET_USER_ACCESS.UID, Integer.valueOf(uid))
+        .set(DATASET_USER_ACCESS.PRIVILEGE, privilege)
+        .execute()
+    }
+    val tables = VersionedResourceTables.DatasetTables
+    def grantsSeenBy(uid: Integer): Seq[PrivilegeEnum] =
+      HubResource
+        .versionedResourcesByIdsQuery(tables, Seq(did), uid)
+        .fetch()
+        .asScala
+        .map(_.get(DATASET_USER_ACCESS.PRIVILEGE))
+        .toSeq
+
+    grantsSeenBy(Integer.valueOf(likerUid)) shouldBe Seq(PrivilegeEnum.READ)
+    // A user with no grant, and an anonymous viewer, see one row with no grant at all.
+    grantsSeenBy(Integer.valueOf(thirdUid)) shouldBe Seq(null)
+    grantsSeenBy(null) shouldBe Seq(null)
+  }
+
   "getTops" should "list the liked public entity under the 'like' key" in {
     recordLikeAction(req, Integer.valueOf(likerUid), UserRequest(wid, Wf), isLike = true)
 
diff --git a/build.sbt b/build.sbt
index bb202e8cbf0..f0f09e2fead 100644
--- a/build.sbt
+++ b/build.sbt
@@ -127,7 +127,10 @@ lazy val Config = (project in file("common/config")).settings(commonModuleSettin
 // on Config to read OTEL_* settings from observability.conf.
 lazy val Observability =
   (project in file("common/observability")).settings(commonModuleSettings).dependsOn(Config)
-lazy val Resource = (project in file("common/resource")).settings(commonModuleSettings).dependsOn(DAO)
+lazy val Resource = (project in file("common/resource"))
+  .settings(commonModuleSettings)
+  .dependsOn(DAO, Auth)
+  .dependsOn(DAO % "test->test") // reuse MockTexeraDB embedded Postgres in tests
 lazy val Auth = (project in file("common/auth"))
   .settings(commonModuleSettings)
   .configs(Test)
diff --git a/common/auth/src/main/scala/org/apache/texera/auth/util/ComputingUnitAccess.scala b/common/auth/src/main/scala/org/apache/texera/auth/util/ComputingUnitAccess.scala
index b999cb48906..8115c0d0019 100644
--- a/common/auth/src/main/scala/org/apache/texera/auth/util/ComputingUnitAccess.scala
+++ b/common/auth/src/main/scala/org/apache/texera/auth/util/ComputingUnitAccess.scala
@@ -48,10 +48,8 @@ object ComputingUnitAccess {
 
     if (record == null) {
       PrivilegeEnum.NONE // no such unit
-    } else if (record.value1().equals(uid)) {
-      PrivilegeEnum.WRITE // owner
     } else {
-      Option(record.value2()).getOrElse(PrivilegeEnum.NONE)
+      OwnerAccess.privilege(record.value1().equals(uid), record.value2())
     }
   }
 }
diff --git a/common/auth/src/main/scala/org/apache/texera/auth/util/OwnerAccess.scala b/common/auth/src/main/scala/org/apache/texera/auth/util/OwnerAccess.scala
new file mode 100644
index 00000000000..45b37dbf612
--- /dev/null
+++ b/common/auth/src/main/scala/org/apache/texera/auth/util/OwnerAccess.scala
@@ -0,0 +1,37 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 org.apache.texera.auth.util
+
+import jakarta.ws.rs.ForbiddenException
+import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
+
+/** The rule every shared resource applies to its owner. */
+object OwnerAccess {
+
+  /** The owner always has WRITE; anyone else has what they were granted, or NONE.
+    * `granted` is by-name, so a lookup passed in is skipped for the owner.
+    */
+  def privilege(isOwner: Boolean, granted: => PrivilegeEnum): PrivilegeEnum =
+    if (isOwner) PrivilegeEnum.WRITE else Option(granted).getOrElse(PrivilegeEnum.NONE)
+
+  val ownerChangeRefused = "You cannot modify the owner's permissions!"
+
+  /** Refuses any change to the owner's own access. */
+  def requireNotOwner(isOwner: Boolean): Unit =
+    if (isOwner) throw new ForbiddenException(ownerChangeRefused)
+}
diff --git a/common/auth/src/test/scala/org/apache/texera/auth/util/OwnerAccessSpec.scala b/common/auth/src/test/scala/org/apache/texera/auth/util/OwnerAccessSpec.scala
new file mode 100644
index 00000000000..656cda38552
--- /dev/null
+++ b/common/auth/src/test/scala/org/apache/texera/auth/util/OwnerAccessSpec.scala
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 org.apache.texera.auth.util
+
+import jakarta.ws.rs.ForbiddenException
+import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class OwnerAccessSpec extends AnyFlatSpec with Matchers {
+
+  "privilege" should "give the owner WRITE whatever they were granted" in {
+    OwnerAccess.privilege(isOwner = true, PrivilegeEnum.READ) shouldBe PrivilegeEnum.WRITE
+    OwnerAccess.privilege(isOwner = true, null) shouldBe PrivilegeEnum.WRITE
+  }
+
+  it should "give anyone else their grant, or NONE without one" in {
+    OwnerAccess.privilege(isOwner = false, PrivilegeEnum.READ) shouldBe PrivilegeEnum.READ
+    OwnerAccess.privilege(isOwner = false, null) shouldBe PrivilegeEnum.NONE
+  }
+
+  it should "not look up the grant for the owner" in {
+    var lookups = 0
+    def lookup(): PrivilegeEnum = { lookups += 1; PrivilegeEnum.READ }
+
+    OwnerAccess.privilege(isOwner = true, lookup())
+    lookups shouldBe 0
+    OwnerAccess.privilege(isOwner = false, lookup())
+    lookups shouldBe 1
+  }
+
+  "requireNotOwner" should "refuse a change to the owner's access with a 403" in {
+    val ex = intercept[ForbiddenException](OwnerAccess.requireNotOwner(isOwner = true))
+    ex.getResponse.getStatus shouldBe 403
+    ex.getMessage shouldBe OwnerAccess.ownerChangeRefused
+  }
+
+  it should "allow a change to anyone else's access" in {
+    noException should be thrownBy OwnerAccess.requireNotOwner(isOwner = false)
+  }
+}
diff --git a/common/resource/build.sbt b/common/resource/build.sbt
index ab974601558..9fe230d1671 100644
--- a/common/resource/build.sbt
+++ b/common/resource/build.sbt
@@ -51,5 +51,6 @@ Compile / scalacOptions ++= Seq(
 libraryDependencies ++= Seq(
   "jakarta.ws.rs" % "jakarta.ws.rs-api" % "3.0.0",           // JAX-RS annotations for shared REST resources
   "jakarta.annotation" % "jakarta.annotation-api" % "2.1.1", // @PermitAll on shared REST resources
-  "org.scalatest" %% "scalatest" % "3.2.15" % Test           // ScalaTest (for unit tests)
+  "org.scalatest" %% "scalatest" % "3.2.15" % Test,          // ScalaTest (for unit tests)
+  "org.glassfish.jersey.core" % "jersey-common" % "3.0.12" % Test // JAX-RS runtime, so tests can build a ForbiddenException
 )
diff --git a/common/resource/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala b/common/resource/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
index b12204877d7..edf5364caf1 100644
--- a/common/resource/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
+++ b/common/resource/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
@@ -21,6 +21,7 @@ package org.apache.texera.service.resource
 
 import jakarta.ws.rs.{BadRequestException, ForbiddenException}
 import jakarta.ws.rs.core.Response
+import org.apache.texera.auth.util.OwnerAccess
 import org.apache.texera.dao.jooq.generated.Tables.USER
 import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
 import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
@@ -69,13 +70,15 @@ object ResourceAccess {
     ).flatMap(record => Option(record.value1()))
       .contains(uid)
 
+  /** The owner always has WRITE, whatever their own access row says. */
   def privilegeOf[R <: Record, A <: Record](
       ctx: DSLContext,
       resource: ResourceTables[R, A],
       id: Integer,
       uid: Integer
   ): PrivilegeEnum =
-    Option(
+    OwnerAccess.privilege(
+      userOwns(ctx, resource, id, uid),
       ctx
         .select(resource.privilegeField)
         .from(resource.accessTable)
@@ -85,7 +88,7 @@ object ResourceAccess {
             .and(resource.accessUidField.eq(uid))
         )
         .fetchOneInto(classOf[PrivilegeEnum])
-    ).getOrElse(PrivilegeEnum.NONE)
+    )
 
   def userHasWriteAccess[R <: Record, A <: Record](
       ctx: DSLContext,
@@ -93,8 +96,7 @@ object ResourceAccess {
       id: Integer,
       uid: Integer
   ): Boolean =
-    userOwns(ctx, resource, id, uid) ||
-      privilegeOf(ctx, resource, id, uid) == PrivilegeEnum.WRITE
+    privilegeOf(ctx, resource, id, uid) == PrivilegeEnum.WRITE
 
   def userHasReadAccess[R <: Record, A <: Record](
       ctx: DSLContext,
@@ -102,9 +104,7 @@ object ResourceAccess {
       id: Integer,
       uid: Integer
   ): Boolean =
-    isPublic(ctx, resource, id) ||
-      userHasWriteAccess(ctx, resource, id, uid) ||
-      privilegeOf(ctx, resource, id, uid) == PrivilegeEnum.READ
+    isPublic(ctx, resource, id) || privilegeOf(ctx, resource, id, uid) != PrivilegeEnum.NONE
 
   /** The owning user, or null when the resource does not exist. */
   def owner[R <: Record, A <: Record](
@@ -174,7 +174,7 @@ object ResourceAccess {
   }
 
   /**
-    * Every resource of this type the user may see: the ones they hold an explicit grant on, plus
+    * Every resource of this type the user may see: the ones they own or hold an explicit grant on, plus
     * every public one, with public entries dropped when they duplicate a granted entry.
     *
     * @param pojoClass  the generated POJO the resource table maps into
@@ -199,18 +199,22 @@ object ResourceAccess {
       .from(
         resource.table
           .leftJoin(resource.accessTable)
-          .on(resource.accessIdField.eq(resource.idField))
+          .on(resource.accessIdField.eq(resource.idField).and(resource.accessUidField.eq(uid)))
           .leftJoin(USER)
           .on(USER.UID.eq(resource.ownerUidField))
       )
-      .where(resource.accessUidField.eq(uid))
+      .where(resource.accessUidField.eq(uid).or(resource.ownerUidField.eq(uid)))
       .fetch()
       .asScala
       .toList
       .flatMap { record =>
         val entity = record.into(resource.table).into(pojoClass)
-        val privilege = record.into(resource.accessTable).get(resource.privilegeField)
         val isOwner = record.into(resource.table).get(resource.ownerUidField) == uid
+        val privilege =
+          OwnerAccess.privilege(
+            isOwner,
+            record.into(resource.accessTable).get(resource.privilegeField)
+          )
         fromGrant(entity, record.into(USER).getEmail, privilege, isOwner)
           .map(entry => (idOf(entity), entry))
       }
@@ -257,6 +261,7 @@ object ResourceAccess {
       throw new BadRequestException(s"No registered user with email $email")
     }
     val granteeUid = grantee.getUid
+    OwnerAccess.requireNotOwner(userOwns(ctx, resource, id, granteeUid))
     val granted = PrivilegeEnum.valueOf(privilege)
 
     ctx
@@ -275,7 +280,8 @@ object ResourceAccess {
   /**
     * Removes the user's explicit grant; a no-op when they hold none.
     *
-    * @throws jakarta.ws.rs.ForbiddenException if the caller cannot modify the resource.
+    * @throws jakarta.ws.rs.ForbiddenException if the caller cannot modify the resource, or the
+    *                                          user is its owner.
     */
   def revoke[R <: Record, A <: Record](
       ctx: DSLContext,
@@ -286,6 +292,7 @@ object ResourceAccess {
   ): Response = {
     requireWriteAccess(ctx, resource, id, requesterUid)
     val granteeUid = new UserDao(ctx.configuration()).fetchOneByEmail(email).getUid
+    OwnerAccess.requireNotOwner(userOwns(ctx, resource, id, granteeUid))
 
     ctx
       .delete(resource.accessTable)
@@ -324,7 +331,7 @@ object ResourceAccess {
     }
 
   /**
-    * Emails of the owners of every resource the caller has an explicit grant on, for the
+    * Emails of the owners of every resource the caller owns or has an explicit grant on, for the
     * owner facet on list pages.
     */
   def ownerEmailsVisibleTo[R <: Record, A <: Record](
@@ -337,9 +344,9 @@ object ResourceAccess {
       .from(USER)
       .join(resource.table)
       .on(resource.ownerUidField.eq(USER.UID))
-      .join(resource.accessTable)
-      .on(resource.accessIdField.eq(resource.idField))
-      .where(resource.accessUidField.eq(uid))
+      .leftJoin(resource.accessTable)
+      .on(resource.accessIdField.eq(resource.idField).and(resource.accessUidField.eq(uid)))
+      .where(resource.accessUidField.eq(uid).or(resource.ownerUidField.eq(uid)))
       .fetchInto(classOf[String])
 
 }
diff --git a/common/resource/src/test/scala/org/apache/texera/service/resource/ResourceAccessSpec.scala b/common/resource/src/test/scala/org/apache/texera/service/resource/ResourceAccessSpec.scala
new file mode 100644
index 00000000000..5a2b72a4be9
--- /dev/null
+++ b/common/resource/src/test/scala/org/apache/texera/service/resource/ResourceAccessSpec.scala
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.texera.service.resource
+
+import jakarta.ws.rs.ForbiddenException
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.DatasetUserAccess.DATASET_USER_ACCESS
+import org.apache.texera.dao.jooq.generated.tables.daos.{DatasetDao, DatasetUserAccessDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{Dataset, DatasetUserAccess, User}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import scala.jdk.CollectionConverters._
+
+/** The owner always has WRITE, whatever their own access row says, and nobody can change that row. */
+class ResourceAccessSpec
+    extends AnyFlatSpec
+    with Matchers
+    with MockTexeraDB
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach {
+
+  private val resource = ResourceTables.Dataset
+
+  private def newUser(name: String): User = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(s"$name@test.com")
+    user.setRole(UserRoleEnum.REGULAR)
+    user
+  }
+
+  private val owner = newUser("owner")
+  private val grantee = newUser("grantee")
+  private val stranger = newUser("stranger")
+
+  private def newDataset(name: String, isPublic: Boolean): Dataset = {
+    val dataset = new Dataset
+    dataset.setName(name)
+    dataset.setRepositoryName(name)
+    dataset.setIsPublic(isPublic)
+    dataset.setIsDownloadable(true)
+    dataset.setDescription("")
+    dataset
+  }
+
+  private val privateDataset = newDataset("private-dataset", isPublic = false)
+  private val publicDataset = newDataset("public-dataset", isPublic = true)
+
+  private def did: Integer = privateDataset.getDid
+
+  private def grantRow(did: Integer, uid: Integer, privilege: PrivilegeEnum): Unit =
+    new DatasetUserAccessDao(getDSLContext.configuration())
+      .insert(new DatasetUserAccess(did, uid, privilege))
+
+  private def rowOf(uid: Integer): Option[PrivilegeEnum] =
+    Option(
+      getDSLContext
+        .select(DATASET_USER_ACCESS.PRIVILEGE)
+        .from(DATASET_USER_ACCESS)
+        .where(DATASET_USER_ACCESS.DID.eq(did).and(DATASET_USER_ACCESS.UID.eq(uid)))
+        .fetchOneInto(classOf[PrivilegeEnum])
+    )
+
+  /** (dataset name, privilege, isOwner) for every dataset `uid` may see. */
+  private def visibleTo(
+      uid: Integer,
+      includePublic: Boolean = true
+  ): List[(String, PrivilegeEnum, Boolean)] =
+    ResourceAccess.listVisible(
+      getDSLContext,
+      resource,
+      uid,
+      classOf[Dataset],
+      (d: Dataset) => d.getDid,
+      includePublic
+    )(
+      (d, _, privilege, isOwner) => Some((d.getName, privilege, isOwner)),
+      (d, _) => Some((d.getName, PrivilegeEnum.READ, false))
+    )
+
+  override protected def beforeAll(): Unit = {
+    super.beforeAll()
+    initializeDBAndReplaceDSLContext()
+    val userDao = new UserDao(getDSLContext.configuration())
+    Seq(owner, grantee, stranger).foreach(userDao.insert(_))
+    privateDataset.setOwnerUid(owner.getUid)
+    publicDataset.setOwnerUid(owner.getUid)
+    new DatasetDao(getDSLContext.configuration()).insert(privateDataset, publicDataset)
+  }
+
+  override protected def beforeEach(): Unit = {
+    super.beforeEach()
+    // Every test starts with no access rows, so the owner has none either.
+    getDSLContext.deleteFrom(DATASET_USER_ACCESS).execute()
+  }
+
+  override protected def afterAll(): Unit = {
+    try shutdownDB()
+    finally super.afterAll()
+  }
+
+  "privilegeOf" should "give the owner WRITE without an access row" in {
+    ResourceAccess.privilegeOf(
+      getDSLContext,
+      resource,
+      did,
+      owner.getUid
+    ) shouldBe PrivilegeEnum.WRITE
+  }
+
+  it should "give the owner WRITE even when their access row says READ" in {
+    grantRow(did, owner.getUid, PrivilegeEnum.READ)
+
+    ResourceAccess.privilegeOf(
+      getDSLContext,
+      resource,
+      did,
+      owner.getUid
+    ) shouldBe PrivilegeEnum.WRITE
+  }
+
+  it should "read a grantee's privilege from their access row, and NONE without one" in {
+    grantRow(did, grantee.getUid, PrivilegeEnum.READ)
+
+    ResourceAccess.privilegeOf(
+      getDSLContext,
+      resource,
+      did,
+      grantee.getUid
+    ) shouldBe PrivilegeEnum.READ
+    ResourceAccess.privilegeOf(
+      getDSLContext,
+      resource,
+      did,
+      stranger.getUid
+    ) shouldBe PrivilegeEnum.NONE
+  }
+
+  "userHasWriteAccess" should "hold for the owner and a WRITE grantee, not a READ grantee" in {
+    grantRow(did, grantee.getUid, PrivilegeEnum.READ)
+
+    ResourceAccess.userHasWriteAccess(getDSLContext, resource, did, owner.getUid) shouldBe true
+    ResourceAccess.userHasWriteAccess(getDSLContext, resource, did, grantee.getUid) shouldBe false
+  }
+
+  "userHasReadAccess" should "hold for the owner, a grantee and anyone on a public dataset, not a stranger" in {
+    grantRow(did, grantee.getUid, PrivilegeEnum.READ)
+
+    ResourceAccess.userHasReadAccess(getDSLContext, resource, did, owner.getUid) shouldBe true
+    ResourceAccess.userHasReadAccess(getDSLContext, resource, did, grantee.getUid) shouldBe true
+    ResourceAccess.userHasReadAccess(getDSLContext, resource, did, stranger.getUid) shouldBe false
+    ResourceAccess.userHasReadAccess(
+      getDSLContext,
+      resource,
+      publicDataset.getDid,
+      stranger.getUid
+    ) shouldBe true
+  }
+
+  "listVisible" should "list the owner's private dataset as WRITE without an access row" in {
+    visibleTo(owner.getUid, includePublic = false) should contain(
+      ("private-dataset", PrivilegeEnum.WRITE, true)
+    )
+  }
+
+  it should "list the owner's dataset as WRITE even when their access row says READ" in {
+    grantRow(did, owner.getUid, PrivilegeEnum.READ)
+
+    visibleTo(owner.getUid, includePublic = false) should contain(
+      ("private-dataset", PrivilegeEnum.WRITE, true)
+    )
+  }
+
+  it should "list a public dataset its owner holds no row on once, as the owner's" in {
+    visibleTo(owner.getUid).filter(_._1 == "public-dataset") shouldBe List(
+      ("public-dataset", PrivilegeEnum.WRITE, true)
+    )
+  }
+
+  it should "list a grantee's dataset with their granted privilege" in {
+    grantRow(did, grantee.getUid, PrivilegeEnum.READ)
+
+    visibleTo(grantee.getUid, includePublic = false) shouldBe List(
+      ("private-dataset", PrivilegeEnum.READ, false)
+    )
+  }
+
+  it should "show a stranger only public datasets" in {
+    visibleTo(stranger.getUid).map(_._1) shouldBe List("public-dataset")
+    visibleTo(stranger.getUid, includePublic = false) shouldBe empty
+  }
+
+  "ownerEmailsVisibleTo" should "include the owner's own email without an access row" in {
+    ResourceAccess
+      .ownerEmailsVisibleTo(getDSLContext, resource, owner.getUid)
+      .asScala shouldBe List(owner.getEmail)
+  }
+
+  it should "include the owner of a dataset shared with the caller, and nothing for a stranger" in {
+    grantRow(did, grantee.getUid, PrivilegeEnum.READ)
+
+    ResourceAccess
+      .ownerEmailsVisibleTo(getDSLContext, resource, grantee.getUid)
+      .asScala shouldBe List(owner.getEmail)
+    ResourceAccess
+      .ownerEmailsVisibleTo(getDSLContext, resource, stranger.getUid)
+      .asScala shouldBe empty
+  }
+
+  "grant" should "refuse to change the owner's privilege, leaving their row as it was" in {
+    grantRow(did, owner.getUid, PrivilegeEnum.WRITE)
+    grantRow(did, grantee.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      ResourceAccess.grant(getDSLContext, resource, did, owner.getEmail, "READ", grantee.getUid)
+    }
+    ex.getResponse.getStatus shouldBe 403
+    rowOf(owner.getUid) shouldBe Some(PrivilegeEnum.WRITE)
+  }
+
+  it should "still grant to anyone else" in {
+    ResourceAccess
+      .grant(getDSLContext, resource, did, grantee.getEmail, "READ", owner.getUid)
+      .getStatus shouldBe 200
+    rowOf(grantee.getUid) shouldBe Some(PrivilegeEnum.READ)
+  }
+
+  "revoke" should "refuse to remove the owner's access row" in {
+    grantRow(did, owner.getUid, PrivilegeEnum.WRITE)
+    grantRow(did, grantee.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      ResourceAccess.revoke(getDSLContext, resource, did, owner.getEmail, grantee.getUid)
+    }
+    ex.getResponse.getStatus shouldBe 403
+    rowOf(owner.getUid) shouldBe Some(PrivilegeEnum.WRITE)
+  }
+
+  it should "still revoke anyone else" in {
+    grantRow(did, grantee.getUid, PrivilegeEnum.READ)
+
+    ResourceAccess
+      .revoke(getDSLContext, resource, did, grantee.getEmail, owner.getUid)
+      .getStatus shouldBe 200
+    rowOf(grantee.getUid) shouldBe None
+  }
+}
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitAccessResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitAccessResource.scala
index be8a498c09e..2faf81265da 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitAccessResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitAccessResource.scala
@@ -23,6 +23,7 @@ import jakarta.annotation.security.RolesAllowed
 import jakarta.ws.rs.core.MediaType
 import jakarta.ws.rs._
 import org.apache.texera.auth.SessionUser
+import org.apache.texera.auth.util.OwnerAccess
 import org.apache.texera.common.config.ComputingUnitConfig
 import org.apache.texera.dao.SqlServer
 import org.apache.texera.dao.SqlServer.withTransaction
@@ -135,10 +136,17 @@ class ComputingUnitAccessResource {
   ): List[AccessEntry] = {
     ensureSharingIsEnabled()
     withTransaction(context) { ctx =>
-      val computingUnitUserAccessDao = new ComputingUnitUserAccessDao(ctx.configuration())
-      computingUnitUserAccessDao
-        .fetchByCuid(cuid)
-        .asScala
+      val unit = new WorkflowComputingUnitDao(ctx.configuration()).fetchOneByCuid(cuid)
+      val grants = new ComputingUnitUserAccessDao(ctx.configuration()).fetchByCuid(cuid).asScala
+      val canRead =
+        unit != null && (unit.getUid.equals(user.getUid) || grants.exists(
+          _.getUid.equals(user.getUid)
+        ))
+      if (!canRead) {
+        throw new ForbiddenException(s"You do not have access to computing unit $cuid")
+      }
+      grants
+        .filterNot(_.getUid.equals(unit.getUid))
         .map(access => {
           val user = userDao.fetchOneByUid(access.getUid)
           AccessEntry(
@@ -165,6 +173,7 @@ class ComputingUnitAccessResource {
     }
 
     val granteeId = resolveUidByEmail(email)
+    OwnerAccess.requireNotOwner(isOwner(cuid, granteeId))
 
     withTransaction(context) { ctx =>
       val computingUnitUserAccessDao = new ComputingUnitUserAccessDao(ctx.configuration())
@@ -192,6 +201,7 @@ class ComputingUnitAccessResource {
     }
 
     val granteeId = resolveUidByEmail(email)
+    OwnerAccess.requireNotOwner(isOwner(cuid, granteeId))
 
     withTransaction(context) { ctx =>
       ctx
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
index a9326e14a44..2f4fe4a428f 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
@@ -28,6 +28,7 @@ import jakarta.ws.rs.core.{MediaType, Response}
 import org.apache.commons.lang3.StringUtils
 import org.apache.texera.auth.JwtAuth.jwtClaims
 import org.apache.texera.auth.{JwtAuth, SessionUser}
+import org.apache.texera.auth.util.ComputingUnitAccess
 import org.apache.texera.common.config.KubernetesConfig.{
   cpuLimitOptions,
   gpuLimitOptions,
@@ -884,22 +885,7 @@ class ComputingUnitManagingResource {
       statusReason = if (isOwner) statusReason else None,
       metrics = ComputingUnitHelpers.getComputingUnitMetrics(unit),
       isOwner = isOwner,
-      accessPrivilege = {
-        val cuAccessDao = new ComputingUnitUserAccessDao(context.configuration())
-        val access = cuAccessDao
-          .fetchByUid(user.getUid)
-          .asScala
-          .find(access => access.getCuid.equals(cuid))
-
-        if (access.isDefined) {
-          access.get.getPrivilege
-        } else if (unit.getUid.equals(user.getUid)) {
-          PrivilegeEnum.WRITE
-        } else {
-          // Default privilege for non-owners without explicit access
-          PrivilegeEnum.NONE
-        }
-      },
+      accessPrivilege = ComputingUnitAccess.getComputingUnitAccess(cuid, user.getUid),
       ownerAvatar,
       ownerUsername
     )
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala
index 1396784f372..02bbbb4864d 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala
@@ -193,6 +193,21 @@ class ComputingUnitAccessResourceSpec
     entries.head.privilege shouldEqual PrivilegeEnum.WRITE
   }
 
+  it should "refuse to change the owner's privilege" in {
+    grantDirectly(granteeUser.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      accessResource.grantAccess(
+        new SessionUser(granteeUser),
+        cuid,
+        ownerUser.getEmail,
+        PrivilegeEnum.READ
+      )
+    }
+    ex.getResponse.getStatus shouldEqual 403
+    ComputingUnitAccessResource.getPrivilege(cuid, ownerUser.getUid) shouldBe null
+  }
+
   it should "reject a caller without write access with a 403" in {
     val ex = intercept[ForbiddenException] {
       accessResource.grantAccess(strangerSession, cuid, granteeUser.getEmail, PrivilegeEnum.READ)
@@ -222,6 +237,17 @@ class ComputingUnitAccessResourceSpec
     ex.getMessage should include("User with the given email does not exist")
   }
 
+  it should "refuse to revoke the owner's access" in {
+    grantDirectly(granteeUser.getUid, PrivilegeEnum.WRITE)
+    grantDirectly(ownerUser.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      accessResource.revokeAccess(new SessionUser(granteeUser), cuid, ownerUser.getEmail)
+    }
+    ex.getResponse.getStatus shouldEqual 403
+    ComputingUnitAccessResource.getPrivilege(cuid, ownerUser.getUid) shouldBe PrivilegeEnum.WRITE
+  }
+
   it should "reject a caller without write access with a 403" in {
     accessResource.grantAccess(ownerSession, cuid, granteeUser.getEmail, PrivilegeEnum.READ)
 
@@ -240,6 +266,34 @@ class ComputingUnitAccessResourceSpec
     accessResource.getComputingUnitAccessList(ownerSession, cuid) shouldBe empty
   }
 
+  it should "leave out a row held by the owner" in {
+    grantDirectly(ownerUser.getUid, PrivilegeEnum.READ)
+    grantDirectly(granteeUser.getUid, PrivilegeEnum.READ)
+
+    accessEmails(cuid) shouldBe List(granteeUser.getEmail)
+  }
+
+  it should "be forbidden for a user who cannot see the unit" in {
+    val ex = intercept[ForbiddenException] {
+      accessResource.getComputingUnitAccessList(strangerSession, cuid)
+    }
+    ex.getResponse.getStatus shouldEqual 403
+  }
+
+  it should "be forbidden for a unit that does not exist" in {
+    assertThrows[ForbiddenException] {
+      accessResource.getComputingUnitAccessList(ownerSession, nonExistentCuid)
+    }
+  }
+
+  it should "be readable by a READ grantee" in {
+    grantDirectly(granteeUser.getUid, PrivilegeEnum.READ)
+
+    accessResource
+      .getComputingUnitAccessList(new SessionUser(granteeUser), cuid)
+      .map(_.email) shouldBe List(granteeUser.getEmail)
+  }
+
   it should "list every grantee with their email, name, and privilege" in {
     grantDirectly(granteeUser.getUid, PrivilegeEnum.READ)
     grantDirectly(strangerUser.getUid, PrivilegeEnum.WRITE)
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
index 34b3c996cb3..2e18cd9ff82 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
@@ -29,6 +29,7 @@ import org.apache.texera.common.config.EnvironmentalVariable
 import org.apache.texera.auth.SessionUser
 import org.apache.texera.common.config.KubernetesConfig.maxNumOfRunningComputingUnitsPerUser
 import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.tables.ComputingUnitUserAccess.COMPUTING_UNIT_USER_ACCESS
 import org.apache.texera.dao.jooq.generated.enums.{
   PrivilegeEnum,
   UserRoleEnum,
@@ -197,6 +198,21 @@ class ComputingUnitManagingResourceSpec
     info.ownerName shouldBe "victim"
   }
 
+  it should "report WRITE for the owner even when an access row says READ" in {
+    insertLocalUnit(cuid = 9510, ownerUid = user.getUid, name = "owner-with-row-cu")
+    grantAccess(9510, granteeUid = user.getUid, PrivilegeEnum.READ)
+    try {
+      resource.getComputingUnitInfo(9510, user).accessPrivilege shouldBe PrivilegeEnum.WRITE
+    } finally {
+      // Kept out of the owner's unit listing, which the tests below count.
+      getDSLContext
+        .deleteFrom(COMPUTING_UNIT_USER_ACCESS)
+        .where(COMPUTING_UNIT_USER_ACCESS.CUID.eq(9510))
+        .execute()
+      new WorkflowComputingUnitDao(getDSLContext.configuration()).deleteById(9510)
+    }
+  }
+
   it should "throw NotFoundException for a nonexistent unit" in {
     a[NotFoundException] should be thrownBy resource.getComputingUnitInfo(99999, user)
   }
diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala
index 709ede7184b..acbeba70793 100644
--- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala
@@ -322,6 +322,30 @@ class DatasetAccessResourceSpec
     userHasReadAccess(getDSLContext, privateDataset.getDid, strangerUser.getUid) shouldBe true
   }
 
+  it should "refuse to change the owner's privilege" in {
+    grantDirectly(privateDataset.getDid, ownerUser.getUid, PrivilegeEnum.WRITE)
+    grantDirectly(privateDataset.getDid, writeGranteeUser.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      accessResource.grantAccess(
+        privateDataset.getDid,
+        ownerUser.getEmail,
+        "READ",
+        writeGranteeSession
+      )
+    }
+    ex.getResponse.getStatus shouldEqual 403
+    getDSLContext
+      .select(DATASET_USER_ACCESS.PRIVILEGE)
+      .from(DATASET_USER_ACCESS)
+      .where(
+        DATASET_USER_ACCESS.DID
+          .eq(privateDataset.getDid)
+          .and(DATASET_USER_ACCESS.UID.eq(ownerUser.getUid))
+      )
+      .fetchOneInto(classOf[PrivilegeEnum]) shouldEqual PrivilegeEnum.WRITE
+  }
+
   it should "be forbidden for a user without write access" in {
     val ex = intercept[ForbiddenException] {
       accessResource.grantAccess(
@@ -422,6 +446,25 @@ class DatasetAccessResourceSpec
     accessList(privateDataset.getDid) shouldBe empty
   }
 
+  it should "refuse to revoke the owner's access" in {
+    grantDirectly(privateDataset.getDid, ownerUser.getUid, PrivilegeEnum.WRITE)
+    grantDirectly(privateDataset.getDid, writeGranteeUser.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      accessResource.revokeAccess(privateDataset.getDid, ownerUser.getEmail, writeGranteeSession)
+    }
+    ex.getResponse.getStatus shouldEqual 403
+    getDSLContext
+      .select(DATASET_USER_ACCESS.PRIVILEGE)
+      .from(DATASET_USER_ACCESS)
+      .where(
+        DATASET_USER_ACCESS.DID
+          .eq(privateDataset.getDid)
+          .and(DATASET_USER_ACCESS.UID.eq(ownerUser.getUid))
+      )
+      .fetchOneInto(classOf[PrivilegeEnum]) shouldEqual PrivilegeEnum.WRITE
+  }
+
   it should "be forbidden for a user without write access" in {
     grantDirectly(privateDataset.getDid, readGranteeUser.getUid, PrivilegeEnum.READ)
 
diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
index 8acf3af06fb..ba2dae5b9a9 100644
--- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
@@ -1069,7 +1069,8 @@ class DatasetResourceSpec
     dataset.setName(repoName)
     dataset.setRepositoryName(repoName)
     dataset.setDescription("list endpoint - orphan DB row")
-    dataset.setOwnerUid(ownerUser.getUid)
+    // Someone else's public dataset, so the caller reaches it only through the public pass.
+    dataset.setOwnerUid(otherAdminUser.getUid)
     dataset.setIsPublic(true)
     dataset.setIsDownloadable(true)
     datasetDao.insert(dataset)
diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/ModelAccessResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/ModelAccessResourceSpec.scala
index a3db382711a..7056793b615 100644
--- a/file-service/src/test/scala/org/apache/texera/service/resource/ModelAccessResourceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/resource/ModelAccessResourceSpec.scala
@@ -336,6 +336,30 @@ class ModelAccessResourceSpec
     userHasReadAccess(getDSLContext, privateModel.getMid, strangerUser.getUid) shouldBe true
   }
 
+  it should "refuse to change the owner's privilege" in {
+    grantDirectly(privateModel.getMid, ownerUser.getUid, PrivilegeEnum.WRITE)
+    grantDirectly(privateModel.getMid, writeGranteeUser.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      accessResource.grantAccess(
+        privateModel.getMid,
+        ownerUser.getEmail,
+        "READ",
+        writeGranteeSession
+      )
+    }
+    ex.getResponse.getStatus shouldEqual 403
+    getDSLContext
+      .select(MODEL_USER_ACCESS.PRIVILEGE)
+      .from(MODEL_USER_ACCESS)
+      .where(
+        MODEL_USER_ACCESS.MID
+          .eq(privateModel.getMid)
+          .and(MODEL_USER_ACCESS.UID.eq(ownerUser.getUid))
+      )
+      .fetchOneInto(classOf[PrivilegeEnum]) shouldEqual PrivilegeEnum.WRITE
+  }
+
   it should "be forbidden for a user without write access" in {
     val ex = intercept[ForbiddenException] {
       accessResource.grantAccess(
@@ -436,6 +460,25 @@ class ModelAccessResourceSpec
     accessList(privateModel.getMid) shouldBe empty
   }
 
+  it should "refuse to revoke the owner's access" in {
+    grantDirectly(privateModel.getMid, ownerUser.getUid, PrivilegeEnum.WRITE)
+    grantDirectly(privateModel.getMid, writeGranteeUser.getUid, PrivilegeEnum.WRITE)
+
+    val ex = intercept[ForbiddenException] {
+      accessResource.revokeAccess(privateModel.getMid, ownerUser.getEmail, writeGranteeSession)
+    }
+    ex.getResponse.getStatus shouldEqual 403
+    getDSLContext
+      .select(MODEL_USER_ACCESS.PRIVILEGE)
+      .from(MODEL_USER_ACCESS)
+      .where(
+        MODEL_USER_ACCESS.MID
+          .eq(privateModel.getMid)
+          .and(MODEL_USER_ACCESS.UID.eq(ownerUser.getUid))
+      )
+      .fetchOneInto(classOf[PrivilegeEnum]) shouldEqual PrivilegeEnum.WRITE
+  }
+
   it should "be forbidden for a user without write access" in {
     grantDirectly(privateModel.getMid, readGranteeUser.getUid, PrivilegeEnum.READ)
 
diff --git a/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.spec.ts b/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.spec.ts
index 72f4d042a3e..bdb60471d7f 100644
--- a/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.spec.ts
+++ b/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.spec.ts
@@ -278,6 +278,12 @@ describe("ResourceRegistryService", () => {
     expect(registry.entryLink(dataset, 99)).toEqual([HUB_DATASET_RESULT_DETAIL, "5"]);
   });
 
+  it("routes the owner to the private page even without an access row", () => {
+    const dataset = entry({ type: EntityType.Dataset, id: 5, ownerId: 42, accessibleUserIds: [] });
+    expect(registry.entryLink(dataset, 42)).toEqual([USER_DATASET, "5"]);
+    expect(registry.entryLink(dataset, 99)).toEqual([HUB_DATASET_RESULT_DETAIL, "5"]);
+  });
+
   it("routes a model the same way, to its own page or to the hub", () => {
     const model = entry({ type: EntityType.Model, id: 9, accessibleUserIds: [42] });
     expect(registry.entryLink(model, 42)).toEqual([USER_MODEL, "9"]);
diff --git a/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.ts b/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.ts
index a65ce89cf34..d096f2b3230 100644
--- a/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.ts
+++ b/frontend/src/app/dashboard/service/user/resource-registry/resource-registry.service.ts
@@ -107,7 +107,8 @@ export class ResourceRegistryService {
     if (descriptor.hubRoute === undefined) {
       return [descriptor.privateRoute, String(entry.id)];
     }
-    const reachableByViewer = currentUid !== undefined && entry.accessibleUserIds.includes(currentUid);
+    const reachableByViewer =
+      currentUid !== undefined && (entry.ownerId === currentUid || entry.accessibleUserIds.includes(currentUid));
     return [reachableByViewer ? descriptor.privateRoute : descriptor.hubRoute, String(entry.id)];
   }
 }