From f44c37f6bae22732c980ecf4bf1bcde45e8b1a7c Mon Sep 17 00:00:00 2001
From: Yang Zhang <yangz75@uci.edu>
Date: Sun, 18 Oct 2026 15:06:22 -0700
Subject: [PATCH 1/3] feat(workflow): match search and listings against the
 copy on show

Search and the listings it feeds were reading the author's live columns,
which for a pinned workflow is the one copy the public cannot open. A
draft would turn up in a public search under a title nobody has seen, and
the card would advertise a name the detail page does not show.

Each filter is now applied to whichever copy the caller may see:
`onVisibleCopy` builds the same filter twice -- over the live columns for
rows the caller was granted access to, over the frozen ones for rows they
reach only because the workflow is public -- and ORs the two. A
disjunction over bare columns rather than a CASE, so each side stays
eligible for its own fulltext index. Unpinned public rows fall back to
the live columns, so a following workflow searches exactly as it does now.

Listings carry two more things from the same query: the frozen name and
description to show a viewer without granted access, and whether the copy
on show is behind the author's working copy. `constructWhereClause` takes
`includePublic` for this; the other builders accept and ignore it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../dashboard/SearchQueryBuilder.scala        |  14 +-
 .../dashboard/UnifiedResourceSchema.scala     |  19 +-
 .../VersionedResourceSearchQueryBuilder.scala |   3 +-
 .../WorkflowSearchQueryBuilder.scala          | 131 +++++++-
 .../resource/dashboard/hub/HubResource.scala  |  22 +-
 .../workflow/WorkflowPublishService.scala     |  61 +++-
 .../user/workflow/WorkflowResource.scala      |   5 +-
 .../dashboard/UnifiedResourceSchemaSpec.scala |  35 +-
 .../WorkflowSearchQueryBuilderSpec.scala      |  88 ++++-
 .../user/workflow/WorkflowPublishSpec.scala   | 308 +++++++++++++++++-
 10 files changed, 647 insertions(+), 39 deletions(-)

diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala
index c132500625a..76c6e7d9763 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/SearchQueryBuilder.scala
@@ -58,7 +58,17 @@ trait SearchQueryBuilder {
       includePublic: Boolean = false
   ): TableLike[_]
 
-  protected def constructWhereClause(uid: Integer, params: SearchQueryParams): Condition
+  /**
+    * @param includePublic whether public resources the user has not been granted access to are in
+    *                      scope. Builders whose content differs between the public and the private
+    *                      view (workflows, whose public copy is a pinned version) need this to know
+    *                      which copy a filter may match against.
+    */
+  protected def constructWhereClause(
+      uid: Integer,
+      params: SearchQueryParams,
+      includePublic: Boolean = false
+  ): Condition
 
   protected def getGroupByFields: Seq[GroupField] = Seq.empty
 
@@ -78,7 +88,7 @@ trait SearchQueryBuilder {
     val query: SelectGroupByStep[Record] = context
       .selectDistinct(mappedResourceSchema.allFields: _*)
       .from(constructFromClause(uid, params, includePublic))
-      .where(constructWhereClause(uid, params))
+      .where(constructWhereClause(uid, params, includePublic))
     val groupByFields = getGroupByFields
     if (groupByFields.nonEmpty) {
       query.groupBy(groupByFields: _*)
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala
index 40171c8f11e..d1144f635fa 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchema.scala
@@ -80,7 +80,14 @@ object UnifiedResourceSchema {
       // mark the row and route it accordingly.
       workflowDefaultView: Field[DefaultViewEnum] = DSL.cast(null, classOf[DefaultViewEnum]),
       modelFramework: Field[String] = DSL.cast(null, classOf[String]),
-      modelFormat: Field[String] = DSL.cast(null, classOf[String])
+      modelFormat: Field[String] = DSL.cast(null, classOf[String]),
+      workflowHasUnpublishedChanges: Field[java.lang.Boolean] =
+        DSL.cast(null, classOf[java.lang.Boolean]),
+      workflowPublishedName: Field[String] = DSL.cast(null, classOf[String]),
+      workflowPublishedDescription: Field[String] = DSL.cast(null, classOf[String]),
+      workflowPublishedDefaultView: Field[DefaultViewEnum] =
+        DSL.cast(null, classOf[DefaultViewEnum]),
+      viewerHasGrantedAccess: Field[java.lang.Boolean] = DSL.cast(null, classOf[java.lang.Boolean])
   ): UnifiedResourceSchema = {
     new UnifiedResourceSchema(
       Seq(
@@ -110,7 +117,15 @@ object UnifiedResourceSchema {
         workflowCoverImage -> workflowCoverImage.as("workflow_cover_image"),
         workflowDefaultView -> workflowDefaultView.as("workflow_default_view"),
         modelFramework -> modelFramework.as("model_framework"),
-        modelFormat -> modelFormat.as("model_format")
+        modelFormat -> modelFormat.as("model_format"),
+        workflowHasUnpublishedChanges -> workflowHasUnpublishedChanges
+          .as("workflow_has_unpublished_changes"),
+        workflowPublishedName -> workflowPublishedName.as("workflow_published_name"),
+        workflowPublishedDescription -> workflowPublishedDescription
+          .as("workflow_published_description"),
+        workflowPublishedDefaultView -> workflowPublishedDefaultView
+          .as("workflow_published_default_view"),
+        viewerHasGrantedAccess -> viewerHasGrantedAccess.as("viewer_has_granted_access")
       )
     )
   }
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..4b693152a1c 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
@@ -110,7 +110,8 @@ abstract class VersionedResourceSearchQueryBuilder[Rec <: Record, P](
 
   override protected def constructWhereClause(
       uid: Integer,
-      params: DashboardResource.SearchQueryParams
+      params: DashboardResource.SearchQueryParams,
+      includePublic: Boolean
   ): Condition = {
     val splitKeywords = params.keywords.asScala
       .flatMap(_.split("[+\\-()<>~*@\"]"))
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala
index 2e08a98d579..ca0e80fb4ea 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilder.scala
@@ -23,9 +23,10 @@ import org.apache.texera.dao.jooq.generated.Tables._
 import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
 import org.apache.texera.web.resource.dashboard.DashboardResource.DashboardClickableFileEntry
 import org.apache.texera.web.resource.dashboard.FulltextSearchQueryUtils._
+import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowPublishService
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.DashboardWorkflow
 import org.jooq.impl.DSL
-import org.jooq.{Condition, GroupField, Record, TableLike}
+import org.jooq.{Condition, Field, GroupField, Record, TableLike}
 
 import scala.jdk.CollectionConverters.CollectionHasAsScala
 import org.apache.texera.dao.jooq.generated.enums.{DefaultViewEnum, PrivilegeEnum}
@@ -53,7 +54,29 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
       ownerId = WORKFLOW_OF_USER.UID,
       userName = USER.NAME,
       workflowCoverImage = DSL.max(WORKFLOW_COVER_IMAGE.IMAGE).as("workflow_cover_image"),
-      workflowDefaultView = WORKFLOW.DEFAULT_VIEW.as("workflow_default_view")
+      workflowDefaultView = WORKFLOW.DEFAULT_VIEW.as("workflow_default_view"),
+      // The isNotNull guard because isDistinctFrom would read a NULL pin as different content, and
+      // every unpinned public workflow would report drift. Aggregated to stay out of the GROUP BY,
+      // which would otherwise group the search by two TEXT columns.
+      workflowHasUnpublishedChanges = DSL
+        .boolOr(
+          WORKFLOW.PUBLISHED_CONTENT.isNotNull
+            .and(WorkflowPublishService.pinDiffersFromWorkingCopy)
+        )
+        .as("workflow_has_unpublished_changes"),
+      // What a viewer without granted access is shown instead of the author's live metadata. NULL
+      // while following, which the reader treats as "show the live values".
+      workflowPublishedName = DSL.max(WORKFLOW.PUBLISHED_NAME).as("workflow_published_name"),
+      workflowPublishedDescription =
+        DSL.max(WORKFLOW.PUBLISHED_DESCRIPTION).as("workflow_published_description"),
+      // The view belongs with the name and the description: a card that opens the frozen copy in
+      // the author's live view would offer a Form View over a graph that carries no form.
+      workflowPublishedDefaultView =
+        DSL.max(WORKFLOW.PUBLISHED_DEFAULT_VIEW).as("workflow_published_default_view"),
+      // The access join is already restricted to the caller, so this needs no user id of its own.
+      viewerHasGrantedAccess = DSL
+        .boolOr(WORKFLOW_USER_ACCESS.UID.isNotNull)
+        .as("viewer_has_granted_access")
     )
   }
 
@@ -88,9 +111,58 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
     baseQuery.where(condition)
   }
 
+  /** Rows the user was granted access to, as opposed to rows they see only because it is public. */
+  private def grantedAccessCondition(uid: Integer): Condition =
+    if (uid == null) DSL.falseCondition()
+    else WORKFLOW_USER_ACCESS.UID.eq(uid)
+
+  /**
+    * The three searchable columns carrying one copy. They travel together because a pin freezes them
+    * together: a filter over content alone would match a pinned workflow on a title no public viewer
+    * has seen.
+    */
+  private case class WorkflowCopy(
+      name: Field[String],
+      description: Field[String],
+      content: Field[String]
+  )
+
+  private val workingCopy =
+    WorkflowCopy(WORKFLOW.NAME, WORKFLOW.DESCRIPTION, WORKFLOW.CONTENT)
+  private val pinnedCopy =
+    WorkflowCopy(
+      WORKFLOW.PUBLISHED_NAME,
+      WORKFLOW.PUBLISHED_DESCRIPTION,
+      WORKFLOW.PUBLISHED_CONTENT
+    )
+
+  /**
+    * Applies a filter to whichever copy the user is allowed to see: one search can return both their
+    * own workflows and public ones, and a pinned public one must not turn up on keywords that exist
+    * only behind the pin. A disjunction of guarded filters over bare columns rather than a CASE, so
+    * each side stays eligible for its own PGroonga index.
+    */
+  private def onVisibleCopy(
+      uid: Integer,
+      includePublic: Boolean
+  )(build: WorkflowCopy => Condition): Condition = {
+    val onWorkingCopy = build(workingCopy).and(grantedAccessCondition(uid))
+    val onPublicCopy = WORKFLOW.IS_PUBLIC
+      .eq(true)
+      .and(
+        // Following leaves the pinned columns NULL, and the public copy is then the working one.
+        build(pinnedCopy)
+          .or(WORKFLOW.PUBLISHED_CONTENT.isNull.and(build(workingCopy)))
+      )
+    if (uid == null) onPublicCopy
+    else if (includePublic) onWorkingCopy.or(onPublicCopy)
+    else onWorkingCopy
+  }
+
   override protected def constructWhereClause(
       uid: Integer,
-      params: DashboardResource.SearchQueryParams
+      params: DashboardResource.SearchQueryParams,
+      includePublic: Boolean
   ): Condition = {
     val splitKeywords = params.keywords.asScala
       .flatMap(_.split("[+\\-()<>~*@\"]"))
@@ -114,13 +186,23 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
       // Apply owner filter
       .and(getContainsFilter(params.owners, USER.EMAIL))
       // Apply operators filter
-      .and(getOperatorsFilter(params.operators, WORKFLOW.CONTENT))
+      .and(
+        if (params.operators.isEmpty) DSL.noCondition()
+        else
+          onVisibleCopy(uid, includePublic)(copy =>
+            getOperatorsFilter(params.operators, copy.content)
+          )
+      )
       // Apply fulltext search filter
       .and(
-        getFullTextSearchFilter(
-          splitKeywords,
-          List(WORKFLOW.NAME, WORKFLOW.DESCRIPTION, WORKFLOW.CONTENT)
-        )
+        if (splitKeywords.isEmpty) DSL.noCondition()
+        else
+          onVisibleCopy(uid, includePublic)(copy =>
+            getFullTextSearchFilter(
+              splitKeywords,
+              List(copy.name, copy.description, copy.content)
+            )
+          )
       )
   }
 
@@ -141,20 +223,37 @@ object WorkflowSearchQueryBuilder extends SearchQueryBuilder {
       uid: Integer,
       record: Record
   ): DashboardResource.DashboardClickableFileEntry = {
+    val workflow = record.into(WORKFLOW).into(classOf[Workflow])
+    // The select lists specific columns, so the POJO built from the record does not carry this one.
+    // Without it the listing forgets the default-view preference on every refresh.
+    workflow.setDefaultView(record.get("workflow_default_view", classOf[DefaultViewEnum]))
+
+    // A viewer here only because the workflow is public sees the pinned name and description, the
+    // same copy the detail page serves -- otherwise a listing would advertise a title that opening it
+    // does not show. Both are NULL while following, which leaves the live values in place.
+    // Unknown counts as not granted: the reverse is the leak.
+    val granted = Option(record.get("viewer_has_granted_access", classOf[java.lang.Boolean]))
+      .exists(_.booleanValue())
+    if (!granted) {
+      Option(record.get("workflow_published_name", classOf[String])).foreach(workflow.setName)
+      Option(record.get("workflow_published_description", classOf[String]))
+        .foreach(workflow.setDescription)
+      Option(record.get("workflow_published_default_view", classOf[DefaultViewEnum]))
+        .foreach(workflow.setDefaultView)
+    }
+
     val dw = DashboardWorkflow(
       record.into(WORKFLOW_OF_USER).getUid == uid,
       Option(record.get(WORKFLOW_USER_ACCESS.PRIVILEGE, classOf[PrivilegeEnum]))
         .map(_.toString)
         .getOrElse(PrivilegeEnum.NONE.toString),
-      record.into(USER).getName, {
-        // The select lists specific columns, so the POJO built from the record does not carry
-        // this one. Without it the listing forgets the default-view preference on every refresh.
-        val w = record.into(WORKFLOW).into(classOf[Workflow])
-        w.setDefaultView(record.get("workflow_default_view", classOf[DefaultViewEnum]))
-        w
-      },
+      record.into(USER).getName,
+      workflow,
       record.into(USER).getUid,
-      Option(record.get("workflow_cover_image", classOf[String]))
+      Option(record.get("workflow_cover_image", classOf[String])),
+      // Null for the resource types that do not define the column at all.
+      Option(record.get("workflow_has_unpublished_changes", classOf[java.lang.Boolean]))
+        .exists(_.booleanValue())
     )
     DashboardClickableFileEntry(SearchQueryBuilder.WORKFLOW_RESOURCE_TYPE, workflow = Some(dw))
   }
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..7b4aa41c2f2 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
@@ -30,6 +30,7 @@ import org.apache.texera.web.resource.dashboard.VersionedResourceTables
 import org.apache.texera.web.resource.dashboard.hub.ActionType.{Clone, Like, Unlike, View}
 import org.apache.texera.web.resource.dashboard.hub.EntityTables._
 import org.apache.texera.web.resource.dashboard.hub.HubResource._
+import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowPublishService
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.{
   DashboardWorkflow,
   baseWorkflowSelect,
@@ -273,7 +274,26 @@ object HubResource {
       )
       .fetch()
 
-    mapWorkflowEntries(records, uid)
+    // The hub is the public shelf, so everything on it is listed as the public sees it: while a pin
+    // is in place, under the name and description frozen with it rather than the author's live ones
+    // -- for the author too, who is looking at the shelf and not at their own dashboard.
+    val entries = mapWorkflowEntries(records, uid)
+    val pinned = WorkflowPublishService.pinnedListingsOf(entries.map(_.workflow.getWid))
+    entries.map { entry =>
+      pinned.get(entry.workflow.getWid) match {
+        case None => entry
+        case Some(listing) =>
+          entry.workflow.setName(listing.name)
+          entry.workflow.setDescription(listing.description)
+          // The view too: a hub card that opened the frozen copy in the author's live view would
+          // offer a Form View over a graph carrying no form.
+          entry.workflow.setDefaultView(listing.defaultView)
+          // Carried like the search listing carries it: a card advertising the pinned copy has to
+          // open that copy, and without this flag the author's own card would open their editor and
+          // show them something else.
+          entry.copy(hasUnpublishedChanges = listing.hasUnpublishedChanges)
+      }
+    }
   }
 
   /**
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
index 9d6a208fcd0..62d15fde230 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
@@ -26,9 +26,10 @@ import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW
 import org.apache.texera.dao.jooq.generated.enums.DefaultViewEnum
 import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowDao
 import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
-import org.jooq.DSLContext
+import org.jooq.{Condition, DSLContext}
 
 import javax.ws.rs.{BadRequestException, NotFoundException}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
 import scala.util.Try
 
 /**
@@ -248,6 +249,64 @@ object WorkflowPublishService extends LazyLogging {
       workflow.getDefaultView
     )
 
+  /**
+    * [[differs]] as a condition, for the listings that ask about many workflows at once: the same
+    * fields, so a card and the share dialog can never disagree about whether edits are held back.
+    * Only meaningful on a row that is pinned -- while following, every frozen column is NULL and
+    * `isDistinctFrom` would read that as drift.
+    */
+  val pinDiffersFromWorkingCopy: Condition =
+    WORKFLOW.PUBLISHED_CONTENT
+      .isDistinctFrom(WORKFLOW.CONTENT)
+      .or(WORKFLOW.PUBLISHED_NAME.isDistinctFrom(WORKFLOW.NAME))
+      .or(WORKFLOW.PUBLISHED_DESCRIPTION.isDistinctFrom(WORKFLOW.DESCRIPTION))
+      .or(WORKFLOW.PUBLISHED_DEFAULT_VIEW.isDistinctFrom(WORKFLOW.DEFAULT_VIEW))
+
+  /**
+    * What a listing needs about one pinned workflow: the frozen name and description it must show
+    * instead of the author's live ones, and whether those live ones have moved on.
+    */
+  case class PinnedListing(
+      name: String,
+      description: String,
+      defaultView: DefaultViewEnum,
+      hasUnpublishedChanges: Boolean
+  )
+
+  /**
+    * The pinned listings among `wids`, keyed by wid. A workflow that follows the author's latest is
+    * simply absent, which leaves its live values in place and its drift flag false.
+    *
+    * Drift is decided in SQL here rather than by [[differs]], because a listing asks about many
+    * workflows at once and none of their contents are worth shipping back to compare in memory.
+    */
+  def pinnedListingsOf(wids: Seq[Integer]): Map[Integer, PinnedListing] =
+    if (wids.isEmpty) Map()
+    else {
+      val drifted = pinDiffersFromWorkingCopy
+      context
+        .select(
+          WORKFLOW.WID,
+          WORKFLOW.PUBLISHED_NAME,
+          WORKFLOW.PUBLISHED_DESCRIPTION,
+          WORKFLOW.PUBLISHED_DEFAULT_VIEW,
+          drifted
+        )
+        .from(WORKFLOW)
+        .where(WORKFLOW.WID.in(wids: _*).and(WORKFLOW.PUBLISHED_CONTENT.isNotNull))
+        .fetch()
+        .asScala
+        .map(row =>
+          row.get(WORKFLOW.WID) -> PinnedListing(
+            row.get(WORKFLOW.PUBLISHED_NAME),
+            row.get(WORKFLOW.PUBLISHED_DESCRIPTION),
+            row.get(WORKFLOW.PUBLISHED_DEFAULT_VIEW),
+            row.get(drifted)
+          )
+        )
+        .toMap
+    }
+
   /** As [[publicCopyOf]], for callers holding only a wid. 404s unless the workflow is public. */
   def publicCopyOf(wid: Integer): PublicCopy = {
     val workflow = requireWorkflow(wid)
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
index 6fbd19bcc2c..74c4b525405 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
@@ -122,7 +122,10 @@ object WorkflowResource {
       ownerName: String,
       workflow: Workflow,
       ownerId: Integer,
-      coverImage: Option[String]
+      coverImage: Option[String],
+      // Behind the author's working copy? Listings use it to open the published preview instead of
+      // the editor -- for the author too, since that is what the entry is showing.
+      hasUnpublishedChanges: Boolean = false
   )
 
   case class WorkflowWithPrivilege(
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala
index 27856595794..72c4d9eadea 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/UnifiedResourceSchemaSpec.scala
@@ -79,9 +79,10 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
 
   // Sentinels for the two slots that have no convenient distinct table
   // column of the right type; every other slot uses a real generated column so
-  // that all 23 originals render differently from one another.
+  // that all 28 originals render differently from one another.
   private val sentinelResourceType: Field[String] = JDSL.inline("s-resource-type")
   private val sentinelStoragePath: Field[String] = JDSL.inline("s-storage-path")
+  private val sentinelGranted: Field[java.lang.Boolean] = JDSL.inline(java.lang.Boolean.TRUE)
 
   private val sentinelSchema: UnifiedResourceSchema = UnifiedResourceSchema(
     resourceType = sentinelResourceType,
@@ -106,7 +107,12 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
     workflowCoverImage = WORKFLOW_COVER_IMAGE.IMAGE,
     workflowDefaultView = WORKFLOW.DEFAULT_VIEW,
     modelFramework = MODEL.FRAMEWORK,
-    modelFormat = MODEL.FORMAT
+    modelFormat = MODEL.FORMAT,
+    workflowHasUnpublishedChanges = WORKFLOW.IS_PUBLIC,
+    workflowPublishedName = WORKFLOW.PUBLISHED_NAME,
+    workflowPublishedDescription = WORKFLOW.PUBLISHED_DESCRIPTION,
+    workflowPublishedDefaultView = WORKFLOW.PUBLISHED_DEFAULT_VIEW,
+    viewerHasGrantedAccess = sentinelGranted
   )
 
   // Expected projection, in order: alias -> the original it must be built from.
@@ -133,13 +139,18 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
     "workflow_cover_image" -> WORKFLOW_COVER_IMAGE.IMAGE,
     "workflow_default_view" -> WORKFLOW.DEFAULT_VIEW,
     "model_framework" -> MODEL.FRAMEWORK,
-    "model_format" -> MODEL.FORMAT
+    "model_format" -> MODEL.FORMAT,
+    "workflow_has_unpublished_changes" -> WORKFLOW.IS_PUBLIC,
+    "workflow_published_name" -> WORKFLOW.PUBLISHED_NAME,
+    "workflow_published_description" -> WORKFLOW.PUBLISHED_DESCRIPTION,
+    "workflow_published_default_view" -> WORKFLOW.PUBLISHED_DEFAULT_VIEW,
+    "viewer_has_granted_access" -> sentinelGranted
   )
 
   // -- apply(): the projection ------------------------------------------------
 
-  "apply" should "expose all 23 slots as aliases, in the order the UNION ALL depends on" in {
-    sentinelSchema.allFields should have size 23
+  "apply" should "expose all 28 slots as aliases, in the order the UNION ALL depends on" in {
+    sentinelSchema.allFields should have size 28
     sentinelSchema.allFields.map(_.getName) shouldBe expectedProjection.map(_._1)
   }
 
@@ -159,7 +170,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
     // about datasets still union with one that does: the column count and
     // types have to line up.
     val defaults = UnifiedResourceSchema()
-    defaults.allFields should have size 23
+    defaults.allFields should have size 28
     val rendered = ctx.renderInlined(JDSL.select(defaults.allFields: _*))
     rendered should include("'' as \"resourceType\"")
     rendered should include("cast(null as timestamp) as \"resourceCreationTime\"")
@@ -194,12 +205,12 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
   }
 
   it should "collapse the all-defaults projection down to one alias per distinct default" in {
-    // 23 slots, but only seven structurally distinct default expressions, so the
+    // 28 slots, but only seven structurally distinct default expressions, so the
     // de-dup collapses the map to seven entries. Worth pinning because it is
     // surprising, and because it is what makes the keep-first rule observable at
-    // all: allFields stays at 23 while the translation map does not.
+    // all: allFields stays at 28 while the translation map does not.
     val defaults = UnifiedResourceSchema()
-    defaults.allFields should have size 23
+    defaults.allFields should have size 28
     translatedAliases(defaults) shouldBe Seq(
       "resourceType", // DSL.inline("")
       "resourceCreationTime", // cast(null as timestamp)
@@ -211,7 +222,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
     )
   }
 
-  it should "keep every distinct original when the caller supplies 23 distinct Fields" in {
+  it should "keep every distinct original when the caller supplies 28 distinct Fields" in {
     // Nothing to collapse here, which is the control case for the two tests
     // above: the shrinkage they observe comes from duplicate originals only.
     translatedAliases(sentinelSchema) shouldBe expectedProjection.map(_._1)
@@ -219,7 +230,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
 
   it should "drop exactly the duplicated slots of the production workflow projection" in {
     val workflowSchema = WorkflowSearchQueryBuilder.mappedResourceSchema
-    workflowSchema.allFields should have size 23
+    workflowSchema.allFields should have size 28
     val aliases = translatedAliases(workflowSchema)
     // `uid` duplicates ownerId (WORKFLOW_OF_USER.UID); the rest are slots the
     // builder left at their default, and the defaults collide by type.
@@ -238,7 +249,7 @@ class UnifiedResourceSchemaSpec extends AnyFlatSpec with Matchers {
 
   "jOOQ Field equality" should "be structural, which is what makes the de-dup collapse anything" in {
     // If jOOQ ever switched to identity equality, translatedFieldSet would keep
-    // all 23 slots and translateRecord would start reading duplicated columns —
+    // all 28 slots and translateRecord would start reading duplicated columns —
     // the tests above would flip, and this one says why.
     JDSL.cast(null, classOf[Integer]) shouldBe JDSL.cast(null, classOf[Integer])
     JDSL.inline("") shouldBe JDSL.inline("")
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala
index f0d8fc63d9a..4b6a8a8937b 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/WorkflowSearchQueryBuilderSpec.scala
@@ -62,6 +62,15 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers {
   // The select lists default_view under its own alias (not carried by the WORKFLOW POJO),
   // and toEntryImpl reads it back by that alias — the record has to carry the column.
   private val defaultViewField = WORKFLOW.DEFAULT_VIEW.as("workflow_default_view")
+  private val driftField =
+    JDSL.field(JDSL.name("workflow_has_unpublished_changes"), classOf[java.lang.Boolean])
+  private val publishedNameField = JDSL.field(JDSL.name("workflow_published_name"), classOf[String])
+  private val publishedDefaultViewField =
+    WORKFLOW.PUBLISHED_DEFAULT_VIEW.as("workflow_published_default_view")
+  private val publishedDescriptionField =
+    JDSL.field(JDSL.name("workflow_published_description"), classOf[String])
+  private val grantedField =
+    JDSL.field(JDSL.name("viewer_has_granted_access"), classOf[java.lang.Boolean])
 
   private val ownerUid: Integer = Integer.valueOf(42)
   private val viewerUid: Integer = Integer.valueOf(43)
@@ -79,7 +88,14 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers {
       uidValue: Integer = ownerUid,
       privilege: PrivilegeEnum = PrivilegeEnum.WRITE,
       cover: String = "cover-b64",
-      defaultView: DefaultViewEnum = DefaultViewEnum.CANVAS
+      defaultView: DefaultViewEnum = DefaultViewEnum.CANVAS,
+      hasUnpublishedChanges: java.lang.Boolean = null,
+      // The rows these tests describe belong to a viewer who was granted access, which is what
+      // leaves the author's own name and description in place.
+      grantedAccess: java.lang.Boolean = java.lang.Boolean.TRUE,
+      publishedName: String = null,
+      publishedDescription: String = null,
+      publishedDefaultView: DefaultViewEnum = null
   ): Record = {
     val record = ctx.newRecord(
       WORKFLOW.WID,
@@ -89,7 +105,12 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers {
       WORKFLOW_USER_ACCESS.PRIVILEGE,
       USER.NAME,
       coverField,
-      defaultViewField
+      defaultViewField,
+      driftField,
+      publishedNameField,
+      publishedDescriptionField,
+      publishedDefaultViewField,
+      grantedField
     )
     record.set(WORKFLOW.WID, wid)
     record.set(WORKFLOW.NAME, "wf-name")
@@ -99,12 +120,75 @@ class WorkflowSearchQueryBuilderSpec extends AnyFlatSpec with Matchers {
     record.set(USER.NAME, "owner-name")
     record.set(coverField, cover)
     record.set(defaultViewField, defaultView)
+    record.set(driftField, hasUnpublishedChanges)
+    record.set(publishedNameField, publishedName)
+    record.set(publishedDescriptionField, publishedDescription)
+    record.set(publishedDefaultViewField, publishedDefaultView)
+    record.set(grantedField, grantedAccess)
     record
   }
 
   private def workflowOf(record: Record, uid: Integer): DashboardWorkflow =
     WorkflowSearchQueryBuilder.toEntryImpl(uid, record).workflow.get
 
+  // -- what a viewer without granted access is shown --------------------------
+
+  "toEntryImpl" should "serve the published name and description to a viewer without granted access" in {
+    // Publishing pins what the public sees, and a listing is where the public meets a workflow, so
+    // the listing has to serve the pinned copy too rather than the author's live metadata.
+    val record = translatedRecord(
+      grantedAccess = java.lang.Boolean.FALSE,
+      publishedName = "as-published",
+      publishedDescription = "described-as-published"
+    )
+
+    val workflow = workflowOf(record, viewerUid).workflow
+
+    workflow.getName shouldBe "as-published"
+    workflow.getDescription shouldBe "described-as-published"
+  }
+
+  it should "serve the pinned view to a viewer without granted access" in {
+    // A card that opened the frozen copy in the author's live view would offer a Form View over a
+    // graph that carries no form; the view travels with the name and the description.
+    val record = translatedRecord(
+      grantedAccess = java.lang.Boolean.FALSE,
+      defaultView = DefaultViewEnum.FORM,
+      publishedName = "as-published",
+      publishedDefaultView = DefaultViewEnum.CANVAS
+    )
+
+    workflowOf(record, viewerUid).workflow.getDefaultView shouldBe DefaultViewEnum.CANVAS
+  }
+
+  it should "leave the author's live name and description for a viewer who was granted access" in {
+    val record = translatedRecord(
+      grantedAccess = java.lang.Boolean.TRUE,
+      publishedName = "as-published",
+      publishedDescription = "described-as-published"
+    )
+
+    val workflow = workflowOf(record, ownerUid).workflow
+
+    workflow.getName shouldBe "wf-name"
+    workflow.getDescription shouldBe "wf-description"
+  }
+
+  it should "treat an unknown access answer as public, which is the safe direction" in {
+    // Showing the published copy to someone who turns out to have access is harmless; the reverse
+    // would hand the author's live metadata to the public.
+    val record = translatedRecord(grantedAccess = null, publishedName = "as-published")
+
+    workflowOf(record, viewerUid).workflow.getName shouldBe "as-published"
+  }
+
+  it should "keep the author's name when a public workflow has nothing pinned" in {
+    // Nothing to substitute, so the row is left as it is rather than blanked.
+    val record = translatedRecord(grantedAccess = java.lang.Boolean.FALSE, publishedName = null)
+
+    workflowOf(record, viewerUid).workflow.getName shouldBe "wf-name"
+  }
+
   // -- privilege fallback -----------------------------------------------------
 
   "toEntryImpl" should "fall back to NONE when the workflow privilege is NULL" in {
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
index 73306d84e37..23b3a5bafe5 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
@@ -29,7 +29,10 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
   WorkflowUserAccessDao
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos.{User, Workflow, WorkflowUserAccess}
+import org.apache.texera.web.resource.dashboard.DashboardResource.SearchQueryParams
+import org.apache.texera.web.resource.dashboard.hub.HubResource
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.WorkflowIDs
+import org.apache.texera.web.resource.dashboard.{DashboardResource, FulltextSearchQueryUtils}
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -45,7 +48,7 @@ import javax.ws.rs.{BadRequestException, ForbiddenException, NotFoundException}
 /**
   * Covers the publish state a workflow can be in -- following the author's latest content, as
   * publishing has always done, or holding a pinned copy of the version the author froze -- and the
-  * read paths that decide which of the two a caller is served.
+  * read paths that decide which of the two a caller is served, listings and search among them.
   */
 class WorkflowPublishSpec
     extends AnyFlatSpec
@@ -85,6 +88,7 @@ class WorkflowPublishSpec
 
   override protected def beforeAll(): Unit = {
     initializeDBAndReplaceDSLContext()
+    FulltextSearchQueryUtils.usePgroonga = false
     val userDao = new UserDao(getDSLContext.configuration())
     userDao.insert(owner)
     userDao.insert(stranger)
@@ -168,6 +172,40 @@ class WorkflowPublishSpec
       )
       .asInstanceOf[HttpServletRequest]
 
+  private def keywords(values: String*): util.ArrayList[String] = {
+    val list = new util.ArrayList[String]()
+    values.foreach(list.add)
+    list
+  }
+
+  private def keywordIds(values: Integer*): util.ArrayList[Integer] = {
+    val list = new util.ArrayList[Integer]()
+    values.foreach(list.add)
+    list
+  }
+
+  /** The wids a search returns, which is all any of these tests asks of one. */
+  private def searchWids(user: SessionUser, params: SearchQueryParams): List[Integer] =
+    DashboardResource
+      .searchAllResources(user, params, includePublic = true)
+      .results
+      .flatMap(_.workflow.map(_.workflow.getWid))
+
+  /** One workflow's listing row, as the given user would see it in a dashboard or on the Hub. */
+  private def listingOf(user: SessionUser, wid: Integer) =
+    DashboardResource
+      .searchAllResources(
+        user,
+        SearchQueryParams(workflowIDs = keywordIds(wid)),
+        includePublic = true
+      )
+      .results
+      .flatMap(_.workflow)
+      .head
+
+  /** Nobody: the hub as an unauthenticated visitor reads it. */
+  private def anonymous: SessionUser = new SessionUser(new User())
+
   private def statusOf(wid: Integer): WorkflowPublishService.PublishStatus =
     workflowResource.getPublishStatus(wid, ownerSession)
 
@@ -945,4 +983,272 @@ class WorkflowPublishSpec
       .getSize(util.Arrays.asList(pinned))
       .get(pinned) shouldBe publishedContent.length
   }
+
+  behavior of "hub listings"
+
+  it should "list a pinned workflow on the hub under the name and description it froze" in {
+    // The hub is the public shelf: everything on it is listed as the public sees it, the author
+    // included. A pin that this listing did not honour would put the author's live title on that
+    // shelf, and would leave it disagreeing with the hub's own search about what a workflow is
+    // called.
+    val wid = createWorkflow("hub_listing_shows_public_copy")
+    publishPinned(wid)
+    relabel(wid, "renamed_after_pinning", "described_after_pinning")
+
+    for (viewer <- Seq(stranger.getUid, owner.getUid)) {
+      val listed = HubResource.fetchDashboardWorkflowsByWids(Seq(wid), viewer).head.workflow
+      listed.getName shouldBe "hub_listing_shows_public_copy"
+      listed.getDescription shouldBe "a workflow"
+    }
+  }
+
+  it should "list a following workflow on the hub under the author's latest name and description" in {
+    val wid = createWorkflow("hub_listing_follows_latest")
+    workflowResource.makePublic(wid, ownerSession)
+    relabel(wid, "renamed_while_following", "described_while_following")
+
+    val listed = HubResource.fetchDashboardWorkflowsByWids(Seq(wid), owner.getUid).head.workflow
+    listed.getName shouldBe "renamed_while_following"
+    listed.getDescription shouldBe "described_while_following"
+  }
+
+  it should "tell the hub listing when the copy on show is behind the author's working copy" in {
+    // The card advertises the pinned copy, so clicking it has to open that copy. Without this flag
+    // the author's own card would take them to their editor and show them something else -- the
+    // same signal the search listing carries, on the query the hub builds for itself.
+    val wid = createWorkflow("hub_listing_reports_drift")
+    publishPinned(wid)
+
+    def listedDrift(): Boolean =
+      HubResource.fetchDashboardWorkflowsByWids(Seq(wid), owner.getUid).head.hasUnpublishedChanges
+
+    listedDrift() shouldBe false
+    edit(wid, editedContent)
+    listedDrift() shouldBe true
+    workflowResource.pinLatest(wid, ownerSession)
+    listedDrift() shouldBe false
+  }
+
+  it should "clone the author's latest name and description while nothing is pinned" in {
+    val wid = createWorkflow("clone_follows_latest_metadata")
+    workflowResource.makePublic(wid, ownerSession)
+    relabel(wid, "renamed_before_clone", "described_before_clone")
+
+    val cloned =
+      workflowDao.fetchOneByWid(workflowResource.cloneWorkflow(wid, strangerSession, fakeRequest()))
+
+    cloned.getName shouldBe "renamed_before_clone_clone"
+    cloned.getDescription shouldBe "described_before_clone"
+  }
+
+  behavior of "search"
+
+  it should "not match a public workflow on anything that exists only in unpublished edits" in {
+    // Both halves of what public search indexes -- the words in the graph, and the operators in it --
+    // have to stop at the pinned copy, or searching would surface drafts nobody can open.
+    val wid = createWorkflow("search_ignores_drafts")
+    publishPinned(wid)
+    edit(
+      wid,
+      """{"operators":[{"operatorType":"SecretDraftOperator"}],"note":"supersecretdraftword"}"""
+    )
+
+    val byKeyword =
+      searchWids(anonymous, SearchQueryParams(keywords = keywords("supersecretdraftword")))
+    val byOperator =
+      searchWids(anonymous, SearchQueryParams(operators = keywords("SecretDraftOperator")))
+
+    byKeyword should not contain wid
+    byOperator should not contain wid
+  }
+
+  it should "match a public workflow on keywords in its published copy" in {
+    val wid = createWorkflow("search_finds_published")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    searchWids(
+      anonymous,
+      SearchQueryParams(keywords = keywords("content_as_published"))
+    ) should contain(wid)
+  }
+
+  it should "match an unpinned public workflow on the author's latest" in {
+    // Following means the public copy is the working copy, so search must reach it through the same
+    // public path that a pinned workflow reaches its frozen copy through.
+    val wid = createWorkflow("search_finds_unpinned_latest")
+    workflowResource.makePublic(wid, ownerSession)
+    edit(wid, """{"operators":[],"note":"unpinnedsearchword"}""")
+
+    searchWids(
+      anonymous,
+      SearchQueryParams(keywords = keywords("unpinnedsearchword"))
+    ) should contain(wid)
+  }
+
+  it should "still match the author's own workflow on their unpublished edits" in {
+    val wid = createWorkflow("search_finds_own_draft")
+    publishPinned(wid)
+    edit(wid, """{"operators":[],"note":"myowndraftword"}""")
+
+    searchWids(
+      ownerSession,
+      SearchQueryParams(keywords = keywords("myowndraftword"))
+    ) should contain(wid)
+  }
+
+  it should "match a pinned workflow on the name and description the public can see" in {
+    // The graph was already matched against the pinned copy; the title and description are on public
+    // show just as much, so matching them against the author's live values gets it wrong in both
+    // directions at once -- findable by a title nobody has seen, unfindable by the one on screen.
+    val wid = createWorkflow("aapublicnamesearch")
+    relabel(wid, "aapublicnamesearch", "aapublicdescriptionsearch")
+    publishPinned(wid)
+    relabel(wid, "zzsecretnamesearch", "zzsecretdescriptionsearch")
+
+    def anonymousHits(word: String): List[Integer] =
+      searchWids(anonymous, SearchQueryParams(keywords = keywords(word)))
+
+    anonymousHits("aapublicnamesearch") should contain(wid)
+    anonymousHits("aapublicdescriptionsearch") should contain(wid)
+    anonymousHits("zzsecretnamesearch") should not contain wid
+    anonymousHits("zzsecretdescriptionsearch") should not contain wid
+  }
+
+  it should "match an unpinned public workflow on its live name and description" in {
+    val wid = createWorkflow("bbfollowingnamesearch")
+    workflowResource.makePublic(wid, ownerSession)
+    relabel(wid, "bbrenamedwhilefollowing", "a workflow")
+
+    searchWids(
+      anonymous,
+      SearchQueryParams(keywords = keywords("bbrenamedwhilefollowing"))
+    ) should contain(wid)
+  }
+
+  it should "still match the author's own workflow on a name only they can see" in {
+    val wid = createWorkflow("ccownnamesearch")
+    publishPinned(wid)
+    relabel(wid, "ccprivaterenamesearch", "a workflow")
+
+    searchWids(
+      ownerSession,
+      SearchQueryParams(keywords = keywords("ccprivaterenamesearch"))
+    ) should contain(wid)
+  }
+
+  it should "show a public viewer the published name and description in a listing" in {
+    // The listing is where people find and click a workflow, so a title that keeps following the
+    // author defeats the freeze exactly as an unfrozen graph would: a report about a title could be
+    // answered by quietly editing the title.
+    val wid = createWorkflow("listing_name_before")
+    publishPinned(wid)
+    relabel(wid, "listing_name_after", "described after publishing")
+
+    def listedAs(session: SessionUser): (String, String) = {
+      val entry = listingOf(session, wid).workflow
+      (entry.getName, entry.getDescription)
+    }
+
+    // The author is not a public viewer of their own workflow.
+    listedAs(ownerSession) shouldBe ("listing_name_after", "described after publishing")
+    // A stranger reaches it only because it is public.
+    listedAs(strangerSession) shouldBe ("listing_name_before", "a workflow")
+    // ...and the listing now agrees with what opening it shows.
+    workflowResource.retrievePublicWorkflow(wid).name shouldBe "listing_name_before"
+  }
+
+  it should "show a public viewer the author's live name while nothing is pinned" in {
+    val wid = createWorkflow("listing_unpinned_before")
+    workflowResource.makePublic(wid, ownerSession)
+    relabel(wid, "listing_unpinned_after", "a workflow")
+
+    listingOf(strangerSession, wid).workflow.getName shouldBe "listing_unpinned_after"
+  }
+
+  it should "show a public viewer the view the copy on show opens in" in {
+    // The Form View's definition rides inside the content, so a card offering a form over a frozen
+    // canvas would open a form the copy does not carry. The view travels with the name and the
+    // description, and for the author's own row it stays theirs.
+    val wid = createWorkflow("listing_freezes_the_view")
+    publishPinned(wid)
+    getDSLContext
+      .update(WORKFLOW)
+      .set(WORKFLOW.DEFAULT_VIEW, DefaultViewEnum.FORM)
+      .where(WORKFLOW.WID.eq(wid))
+      .execute()
+
+    listingOf(strangerSession, wid).workflow.getDefaultView shouldBe DefaultViewEnum.CANVAS
+    listingOf(ownerSession, wid).workflow.getDefaultView shouldBe DefaultViewEnum.FORM
+  }
+
+  it should "show a collaborator the author's live name in a listing" in {
+    val wid = createWorkflow("listing_for_collaborator")
+    publishPinned(wid)
+    grantAccess(wid, PrivilegeEnum.READ)
+
+    try {
+      relabel(wid, "renamed_after_sharing", "a workflow")
+      listingOf(strangerSession, wid).workflow.getName shouldBe "renamed_after_sharing"
+    } finally revokeAccess(wid)
+  }
+
+  it should "leave a private workflow's own listing untouched" in {
+    val wid = createWorkflow("listing_private")
+    relabel(wid, "private_renamed", "a workflow")
+
+    listingOf(ownerSession, wid).workflow.getName shouldBe "private_renamed"
+  }
+
+  it should "tell listings when the copy on show is behind the author's working copy" in {
+    // What sends the author to the published preview instead of their editor when they click their
+    // own workflow in the hub: the entry is advertising the pinned version, not what they are editing.
+    val wid = createWorkflow("listing_reports_drift")
+    publishPinned(wid)
+
+    def drifted(): Boolean = listingOf(ownerSession, wid).hasUnpublishedChanges
+
+    drifted() shouldBe false
+    edit(wid, editedContent)
+    drifted() shouldBe true
+    workflowResource.pinLatest(wid, ownerSession)
+    drifted() shouldBe false
+  }
+
+  it should "report the same drift to a listing as to the share dialog" in {
+    // Three places answer "is the public behind?": the share dialog in Scala, the search projection
+    // and the hub listing in SQL. A rename or a change of view exercises the fields most easily left
+    // out of one of them, and a card that disagrees with the dialog is a card the author distrusts.
+    val wid = createWorkflow("listing_matches_dialog")
+    publishPinned(wid)
+
+    relabel(wid, "renamed_after_pinning", "a workflow")
+    listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe statusOf(wid).hasUnpublishedChanges
+    listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe true
+
+    workflowResource.pinLatest(wid, ownerSession)
+    getDSLContext
+      .update(WORKFLOW)
+      .set(WORKFLOW.DEFAULT_VIEW, DefaultViewEnum.FORM)
+      .where(WORKFLOW.WID.eq(wid))
+      .execute()
+
+    listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe statusOf(wid).hasUnpublishedChanges
+    listingOf(ownerSession, wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "never report drift for a workflow with nothing frozen" in {
+    // Drift is "what the public sees is not what you have", so it can only be true of a workflow that
+    // has a frozen copy at all. Private and public-but-following both have none.
+    def driftOf(wid: Integer): Boolean = listingOf(ownerSession, wid).hasUnpublishedChanges
+
+    val priv = createWorkflow("listing_ignores_private")
+    edit(priv, editedContent)
+    driftOf(priv) shouldBe false
+
+    val following = createWorkflow("listing_ignores_unpinned")
+    workflowResource.makePublic(following, ownerSession)
+    edit(following, editedContent)
+    driftOf(following) shouldBe false
+  }
 }

From 43494d3ffafc774ec5bdff7bfc35cc1125dc221a Mon Sep 17 00:00:00 2001
From: Yang Zhang <yangz75@uci.edu>
Date: Sat, 22 Aug 2026 14:59:57 -0700
Subject: [PATCH 2/3] feat(workflow): mark the pinned version in the revision
 history

The author can pin a version, and their working copy then moves on. This
gives them a way back to what the public is seeing: pinning leaves an
anchor in the revision history they already use, and the panel marks it
as the one currently public, so restoring the published version is the
restore they already know.

The anchor is a version row whose delta is the identity patch, so
replaying it returns exactly what was pinned however many edits pile up
after. An existing row cannot stand in: a version row replays to the
content as it was *before* the change it records. Pinning content that is
already the pinned one reuses its anchor rather than adding a twin, and
the read that decides takes a row lock so two pins racing cannot both
insert.

Two consequences the anchor forces:

- It must not start the version panel's aggregation window. It lands
  seconds after the save it freezes, and the panel folds close-together
  versions into the newest, which would hide the author's own save behind
  a row they never made.
- The revision history is now readable only with granted access, or while
  nothing is pinned. Replaying a version folds deltas back from the
  author's current content, so listing versions of a pinned workflow would
  hand a public viewer the very edits the pin is holding back.

`publish-status` carries the pinned version's date, read from the version
row so the dialog and the panel print one value rather than two clocks'.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../workflow/WorkflowPublishService.scala     |  70 ++++++++--
 .../user/workflow/WorkflowResource.scala      |   6 +-
 .../workflow/WorkflowVersionResource.scala    |  48 ++++---
 .../user/workflow/WorkflowPublishSpec.scala   | 129 ++++++++++++++++--
 .../WorkflowVersionResourceSpec.scala         |  32 ++++-
 5 files changed, 246 insertions(+), 39 deletions(-)

diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
index 62d15fde230..61483f25600 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
@@ -22,7 +22,7 @@ package org.apache.texera.web.resource.dashboard.user.workflow
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW
+import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW, WORKFLOW_VERSION}
 import org.apache.texera.dao.jooq.generated.enums.DefaultViewEnum
 import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowDao
 import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
@@ -30,6 +30,8 @@ import org.jooq.{Condition, DSLContext}
 
 import javax.ws.rs.{BadRequestException, NotFoundException}
 import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+import java.sql.Timestamp
 import scala.util.Try
 
 /**
@@ -54,10 +56,14 @@ object WorkflowPublishService extends LazyLogging {
     * What the share dialog asks about: whether the workflow is public, whether a version is pinned,
     * and whether that pin is holding edits back -- the last is true when pinning again would publish
     * something, and always false while following.
+    *
+    * @param pinnedVersionTime the date naming the pinned version in both the dialog and the
+    *                           revision panel.
     */
   case class PublishStatus(
       isPublished: Boolean,
       isPinned: Boolean,
+      pinnedVersionTime: Option[Timestamp],
       hasUnpublishedChanges: Boolean
   )
 
@@ -112,11 +118,14 @@ object WorkflowPublishService extends LazyLogging {
     * two statements, so a `/private` landing between them would otherwise be undone here, putting a
     * workflow the author had just taken down back on show under a frozen copy.
     *
-    * @return how many rows it matched -- zero for a workflow that is missing or not published.
+    * @return how many rows it matched -- zero for a workflow that is missing or not published.    *
+    * `versionId` names the row in the revision history that replays to this copy, so the history can
+    * mark which version is the published one.
     */
-  private def writePin(wid: Integer): Int =
-    context
+  private def writePin(ctx: DSLContext, wid: Integer, versionId: Integer): Unit =
+    ctx
       .update(WORKFLOW)
+      .set(WORKFLOW.PUBLISHED_VERSION_ID, versionId)
       .set(WORKFLOW.PUBLISHED_CONTENT, WORKFLOW.CONTENT)
       .set(WORKFLOW.PUBLISHED_NAME, WORKFLOW.NAME)
       .set(WORKFLOW.PUBLISHED_DESCRIPTION, WORKFLOW.DESCRIPTION)
@@ -129,9 +138,6 @@ object WorkflowPublishService extends LazyLogging {
     * row only with every frozen column set on a public workflow, or with every one of them NULL, so
     * clearing them one at a time -- or clearing them after `is_public` -- would be rejected.
     *
-    * `published_version_id` is named by that constraint as well but is not touched here, for the
-    * same reason [[writePin]] does not set it: nothing writes it yet, so it is NULL on every row.
-    *
     * Dropping a pin is offered on a published workflow only, so that condition travels in the same
     * statement: a `/private` landing between the guard and here cannot be answered with "unpinned,
     * following latest" about a workflow that is now down. Unpublishing carries no such condition --
@@ -142,6 +148,7 @@ object WorkflowPublishService extends LazyLogging {
   private def clearPin(wid: Integer, alsoUnpublish: Boolean = false): Int = {
     val cleared = context
       .update(WORKFLOW)
+      .set(WORKFLOW.PUBLISHED_VERSION_ID, null.asInstanceOf[Integer])
       .set(WORKFLOW.PUBLISHED_CONTENT, null.asInstanceOf[String])
       .set(WORKFLOW.PUBLISHED_NAME, null.asInstanceOf[String])
       .set(WORKFLOW.PUBLISHED_DESCRIPTION, null.asInstanceOf[String])
@@ -161,9 +168,35 @@ object WorkflowPublishService extends LazyLogging {
     * matching means one of two things, and the caller is told which.
     */
   def pinLatest(wid: Integer): PublishStatus = {
-    if (writePin(wid) == 0) {
-      requireWorkflow(wid) // 404s if it is gone
-      throw new BadRequestException(s"Workflow $wid is not published")
+    context.transaction { txConfig =>
+      val ctx = org.jooq.impl.DSL.using(txConfig)
+      // Locked, not merely read: two pins racing would both read before either wrote, and both
+      // insert an anchor. The lock is also what holds the publication check below: a `/private`
+      // arriving after it has to wait for this transaction rather than slipping under it.
+      val workflow = ctx
+        .selectFrom(WORKFLOW)
+        .where(WORKFLOW.WID.eq(wid))
+        .forUpdate()
+        .fetchOneInto(classOf[Workflow])
+      if (workflow == null) {
+        throw new NotFoundException(s"Workflow $wid not found")
+      }
+      // Before the anchor, not after: a pin that cannot land must not leave a version row behind.
+      if (!workflow.getIsPublic) {
+        throw new BadRequestException(s"Workflow $wid is not published")
+      }
+
+      // An anchor in the revision history for the copy being pinned: its delta is the identity patch,
+      // so replaying this row returns what was published however many edits pile up later. An
+      // existing version row cannot stand in, since one replays to the content as it was *before*
+      // the change it records. Pinning again unchanged reuses the anchor rather than adding a twin.
+      val anchorVid = Option(workflow.getPublishedVersionId)
+        .filter(_ =>
+          Option(workflow.getPublishedContent).exists(sameContent(_, workflow.getContent))
+        )
+        .getOrElse(WorkflowVersionResource.insertNewVersion(wid, ctx = ctx).getVid)
+
+      writePin(ctx, wid, anchorVid)
     }
     logger.info(s"Workflow $wid pinned to its latest content")
     statusOf(wid)
@@ -193,6 +226,22 @@ object WorkflowPublishService extends LazyLogging {
     logger.info(s"Workflow $wid unpublished")
   }
 
+  /** Read from the version row, so the dialog and the revision panel print one date rather than two. */
+  private def pinnedVersionTimeOf(versionId: Integer): Option[Timestamp] =
+    Option(
+      context
+        .select(WORKFLOW_VERSION.CREATION_TIME)
+        .from(WORKFLOW_VERSION)
+        .where(WORKFLOW_VERSION.VID.eq(versionId))
+        .fetchOneInto(classOf[Timestamp])
+    )
+
+  /** The date a public viewer should see: that of the version on show, not of an edit they cannot. */
+  def publicModifiedTime(workflow: Workflow): Timestamp =
+    Option(workflow.getPublishedVersionId)
+      .flatMap(pinnedVersionTimeOf)
+      .getOrElse(workflow.getLastModifiedTime)
+
   /** Whether a version is pinned, and whether it is holding edits back. */
   def statusOf(wid: Integer): PublishStatus = {
     val workflow = requireWorkflow(wid)
@@ -200,6 +249,7 @@ object WorkflowPublishService extends LazyLogging {
     PublishStatus(
       isPublished = workflow.getIsPublic,
       isPinned = pinned,
+      pinnedVersionTime = Option(workflow.getPublishedVersionId).flatMap(pinnedVersionTimeOf),
       // Literally "what the public sees is not what you have": whatever [[publicCopyOf]] freezes is
       // what this compares, on values rather than version ids, so an edit and its undo cancel out.
       hasUnpublishedChanges = differs(publicCopyOf(workflow), workingCopyOf(workflow))
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
index 74c4b525405..e5d71281a67 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
@@ -964,14 +964,16 @@ class WorkflowResource extends LazyLogging {
       .fetchOne()
     // Name and description come from the public copy for the same reason as the content: a pin has
     // to hold everything on show.
-    val publicCopy = WorkflowPublishService.publicCopyOf(workflow.into(classOf[Workflow]))
+    val stored = workflow.into(classOf[Workflow])
+    val publicCopy = WorkflowPublishService.publicCopyOf(stored)
     WorkflowWithPrivilege(
       publicCopy.name,
       publicCopy.description,
       workflow.getWid,
       publicCopy.content,
       workflow.getCreationTime,
-      workflow.getLastModifiedTime,
+      // Dated by the version on show, not by the author's most recent private edit.
+      WorkflowPublishService.publicModifiedTime(stored),
       workflow.getIsPublic,
       readonly = true,
       // The view freezes with the copy: the form's definition lives inside the content.
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala
index c218a1d4ef8..faefabb6393 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResource.scala
@@ -127,11 +127,15 @@ object WorkflowVersionResource {
     *
     * @param wid
     */
-  def insertNewVersion(wid: Integer, content: String = "[]"): WorkflowVersion = {
+  def insertNewVersion(
+      wid: Integer,
+      content: String = "[]",
+      ctx: DSLContext = context
+  ): WorkflowVersion = {
     val workflowVersion = new WorkflowVersion()
     workflowVersion.setContent(content)
     workflowVersion.setWid(wid)
-    workflowVersionDao.insert(workflowVersion)
+    new WorkflowVersionDao(ctx.configuration).insert(workflowVersion)
     workflowVersion
   }
 
@@ -195,39 +199,45 @@ object WorkflowVersionResource {
     * @return
     */
   private def encodeVersionImportance(
-      currentVersions: List[WorkflowVersion]
+      currentVersions: List[WorkflowVersion],
+      publicVersionId: Option[Integer]
   ): List[VersionEntry] = {
     var impEncodedVersions: List[VersionEntry] = List()
 
+    // A pin's anchor is not a version the author made: it carries the identity patch and appears the
+    // moment they pin. It is always shown, so they can find and restore what the public has, but it
+    // must not start the aggregation window -- the save it was pinned from is seconds older, and
+    // would otherwise be folded into it and disappear from the panel.
+    def isAnchor(version: WorkflowVersion): Boolean = publicVersionId.contains(version.getVid)
+
     val lastVersion = currentVersions.head
-    var lastVersionTime = lastVersion.getCreationTime
+    var lastVersionTime: Option[Timestamp] =
+      if (isAnchor(lastVersion)) None else Some(lastVersion.getCreationTime)
     impEncodedVersions = impEncodedVersions :+ VersionEntry(
       lastVersion.getVid,
       lastVersion.getCreationTime,
       lastVersion.getContent,
-      true
-    ) // the first (latest)
-    // version is important even if it is positional
+      true,
+      isAnchor(lastVersion)
+    ) // the first (latest) version is important even if it is positional
     var versionImportance: Boolean = true
     for (version <- currentVersions.tail) {
-      if (
-        isWithinTimeLimit(
-          lastVersionTime,
-          version.getCreationTime
-        )
-      ) {
+      if (isAnchor(version)) {
+        versionImportance = true
+      } else if (lastVersionTime.exists(isWithinTimeLimit(_, version.getCreationTime))) {
         versionImportance = false
       } // try reducing unnecessary check of positional versions
       // because parsing the Json string is expensive
       else {
-        lastVersionTime = version.getCreationTime
+        lastVersionTime = Some(version.getCreationTime)
         versionImportance = isVersionImportant(version.getContent)
       }
       impEncodedVersions = impEncodedVersions :+ VersionEntry(
         version.getVid,
         version.getCreationTime,
         version.getContent,
-        versionImportance
+        versionImportance,
+        isAnchor(version)
       )
     }
     impEncodedVersions
@@ -326,7 +336,10 @@ object WorkflowVersionResource {
       vId: Integer,
       creationTime: Timestamp,
       content: String,
-      importance: Boolean
+      importance: Boolean,
+      // True for the one version the Hub is serving right now, so the author can tell at a glance
+      // where the public copy sits relative to what they are editing.
+      isCurrentlyPublic: Boolean
   )
 
 }
@@ -360,7 +373,8 @@ class WorkflowVersionResource {
           .orderBy(WORKFLOW_VERSION.CREATION_TIME.desc())
           .fetchInto(classOf[WorkflowVersion])
           .asScala
-          .toList
+          .toList,
+        Option(workflowDao.fetchOneByWid(wid)).flatMap(w => Option(w.getPublishedVersionId))
       )
     }
   }
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
index 23b3a5bafe5..9ba7cbcb0a8 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
@@ -48,7 +48,8 @@ import javax.ws.rs.{BadRequestException, ForbiddenException, NotFoundException}
 /**
   * Covers the publish state a workflow can be in -- following the author's latest content, as
   * publishing has always done, or holding a pinned copy of the version the author froze -- and the
-  * read paths that decide which of the two a caller is served, listings and search among them.
+  * read paths that decide which of the two a caller is served, listings and search among them, and
+  * the anchor a pin leaves in the revision history.
   */
 class WorkflowPublishSpec
     extends AnyFlatSpec
@@ -144,14 +145,17 @@ class WorkflowPublishSpec
   private val aWorkflowUpdate =
     """(?is)\s*update\s+(?:\S*\.)?["`\[]?workflow["`\]]?\s+set\b.*""".r
 
-  private def interleaving(interleaved: () => Unit)(act: => Unit): Unit = {
+  private def interleaving(
+      interleaved: () => Unit,
+      at: String => Boolean = aWorkflowUpdate.matches
+  )(act: => Unit): Unit = {
     var pending = true
     val configuration = getDSLContext.configuration().asInstanceOf[DefaultConfiguration]
     val previousListeners = configuration.executeListenerProviders()
     configuration.set(new DefaultExecuteListenerProvider(new ExecuteListener {
       override def executeStart(ctx: ExecuteContext): Unit = {
-        val sql = Option(ctx.sql()).getOrElse("")
-        if (pending && aWorkflowUpdate.matches(sql)) {
+        val sql = Option(ctx.sql()).getOrElse("").toLowerCase
+        if (pending && at(sql)) {
           pending = false
           interleaved()
         }
@@ -160,8 +164,9 @@ class WorkflowPublishSpec
     try act
     finally configuration.set(previousListeners: _*)
     withClue("nothing was interleaved, so this proves nothing: ") { pending shouldBe false }
-  } /** Clone records the caller's IP, and that is the only thing it wants from the request.
-    */
+  }
+
+  /** Clone records the caller's IP, and that is the only thing it wants from the request. */
   private def fakeRequest(): HttpServletRequest =
     Proxy
       .newProxyInstance(
@@ -330,10 +335,14 @@ class WorkflowPublishSpec
     // A pin that read the row and wrote what it had read would freeze the version before a save
     // landing in that window -- and the author, who had just pinned, would be told they have
     // unpublished changes. Each column is copied from its own row instead, so there is no window.
+    //
+    // Interleaved on the way in rather than mid-transaction: pinning now locks the row first, so a
+    // save arriving after that waits for the pin to finish -- which is the point of the lock. This
+    // is the last moment a save can still get in front of it.
     val wid = createWorkflow("pin_takes_the_row_as_it_stands")
     workflowResource.makePublic(wid, ownerSession)
 
-    interleaving(() => edit(wid, editedContent)) {
+    interleaving(() => edit(wid, editedContent), at = _.contains("for update")) {
       workflowResource.pinLatest(wid, ownerSession)
     }
 
@@ -345,11 +354,14 @@ class WorkflowPublishSpec
     // The "is it published?" check and the pin are two statements. An unpublish landing between
     // them used to be undone by the pin's own `is_public = true`, putting a workflow the author had
     // just taken down back on the Hub -- and pinned, so the copy they removed is what the public
-    // reads.
+    // reads. Interleaved on the way in: once the pin has the row's lock, an unpublish waits for it.
     val wid = createWorkflow("pin_cannot_republish")
     publishPinned(wid)
 
-    interleaving(() => workflowResource.makePrivate(wid, ownerSession)) {
+    interleaving(
+      () => workflowResource.makePrivate(wid, ownerSession),
+      at = _.contains("for update")
+    ) {
       a[BadRequestException] should be thrownBy workflowResource.pinLatest(wid, ownerSession)
     }
 
@@ -1251,4 +1263,103 @@ class WorkflowPublishSpec
     edit(following, editedContent)
     driftOf(following) shouldBe false
   }
+
+  behavior of "the revision history"
+
+  it should "leave an anchor in the revision history the author can identify" in {
+    val wid = createWorkflow("publish_leaves_anchor")
+    publishPinned(wid)
+
+    val marked = versionResource
+      .retrieveVersionsOfWorkflow(wid, ownerSession)
+      .filter(_.isCurrentlyPublic)
+    marked should have size 1
+    marked.head.vId shouldBe workflowDao.fetchOneByWid(wid).getPublishedVersionId
+    // Both collapsing rules would otherwise hide it, and a hidden anchor is useless to the author.
+    marked.head.importance shouldBe true
+    // The revision panel prints this row's creation time and the share dialog prints the pinned
+    // version's date. They are one row, so they have to be one value: the same version dated a
+    // second apart in two panels reads as two versions.
+    marked.head.creationTime shouldBe statusOf(wid).pinnedVersionTime.get
+  }
+
+  it should "keep the save a pin was taken from visible in the revision panel" in {
+    // The anchor lands seconds after the save it freezes, and the panel folds versions that land
+    // close together into the newest of them. Letting the anchor be that newest one would hide the
+    // author's own save behind a row they never made.
+    val wid = createWorkflow("pin_does_not_swallow_the_save")
+    edit(wid, editedContent)
+    publishPinned(wid)
+
+    val shown = versionResource
+      .retrieveVersionsOfWorkflow(wid, ownerSession)
+      .filter(_.importance)
+    // The anchor, and the save it was taken from.
+    shown.count(_.isCurrentlyPublic) shouldBe 1
+    shown.size should be >= 2
+  }
+
+  it should "move the mark when the pin moves, leaving the old anchor unmarked" in {
+    val wid = createWorkflow("only_the_live_one_is_marked")
+    publishPinned(wid)
+    val first = workflowDao.fetchOneByWid(wid).getPublishedVersionId
+    edit(wid, editedContent)
+    workflowResource.pinLatest(wid, ownerSession)
+    val second = workflowDao.fetchOneByWid(wid).getPublishedVersionId
+
+    second should not be first
+    versionResource
+      .retrieveVersionsOfWorkflow(wid, ownerSession)
+      .filter(_.isCurrentlyPublic)
+      .map(_.vId) shouldBe List(second)
+  }
+
+  it should "replay the anchor to exactly what was published, however much is edited after" in {
+    val wid = createWorkflow("anchor_replays_to_published")
+    publishPinned(wid)
+    val anchor = workflowDao.fetchOneByWid(wid).getPublishedVersionId
+
+    edit(wid, editedContent)
+    edit(wid, """{"operators":[],"note":"later_still"}""")
+
+    versionResource
+      .retrieveWorkflowVersion(wid, anchor, ownerSession)
+      .getContent shouldBe publishedContent
+  }
+
+  it should "not record a second anchor when pinning again without having edited" in {
+    // Otherwise a repeated click would litter the author's revision panel with rows that all replay
+    // to the same graph.
+    val wid = createWorkflow("republish_without_edits")
+    val before = versionResource.retrieveVersionsOfWorkflow(wid, ownerSession).size
+    publishPinned(wid)
+    val anchor = workflowDao.fetchOneByWid(wid).getPublishedVersionId
+    workflowResource.pinLatest(wid, ownerSession)
+    workflowResource.pinLatest(wid, ownerSession)
+
+    workflowDao.fetchOneByWid(wid).getPublishedVersionId shouldBe anchor
+    versionResource.retrieveVersionsOfWorkflow(wid, ownerSession) should have size (before + 1)
+  }
+
+  it should "keep the version the pin was taken from after the pin is dropped" in {
+    val wid = createWorkflow("unpin_keeps_history")
+    publishPinned(wid)
+
+    workflowResource.unpin(wid, ownerSession)
+
+    versionResource.retrieveVersionsOfWorkflow(wid, ownerSession) should not be empty
+    workflowDao.fetchOneByWid(wid).getPublishedVersionId shouldBe null
+  }
+
+  it should "date the public view by the version on show" in {
+    // A public viewer is told when what they are looking at was published, not when the author last
+    // touched a copy they cannot see.
+    val wid = createWorkflow("public_view_is_dated_by_the_pin")
+    publishPinned(wid)
+    val pinnedAt = statusOf(wid).pinnedVersionTime.get
+
+    edit(wid, editedContent)
+
+    workflowResource.retrievePublicWorkflow(wid).lastModifiedTime shouldBe pinnedAt
+  }
 }
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala
index 91b0b1c1be0..35475ec9ca3 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala
@@ -551,10 +551,40 @@ class WorkflowVersionResourceSpec
         version(3, 0L, positional), // latest — important regardless of content
         version(2, TimeUnit.SECONDS.toMillis(1), meaningful), // within the aggregate window
         version(1, TimeUnit.DAYS.toMillis(30), meaningful) // outside it, judged on content
-      )
+      ),
+      Option.empty[Integer]
     )
 
     encoded.map(_.vId) shouldBe List(3, 2, 1)
     encoded.map(_.importance) shouldBe List(true, false, true)
+    encoded.map(_.isCurrentlyPublic) shouldBe List(false, false, false)
+  }
+
+  it should "keep the version on public show visible even when the rules would collapse it" in {
+    val base = 1_700_000_000_000L
+    def version(vid: Int, offsetMillis: Long, content: String): WorkflowVersion = {
+      val v = new WorkflowVersion
+      v.setVid(vid)
+      v.setWid(testWorkflowWid)
+      v.setContent(content)
+      v.setCreationTime(new Timestamp(base - offsetMillis))
+      v
+    }
+
+    val positional = """[{"op":"replace","path":"/operatorPositions/op-1/x","value":5}]"""
+
+    // Version 2 lands inside the aggregation window and carries a positional-only
+    // patch, so both collapsing rules would hide it. Being the published one wins.
+    val encoded = WorkflowVersionResource invokePrivate encodeVersionImportance(
+      List(
+        version(3, 0L, positional),
+        version(2, TimeUnit.SECONDS.toMillis(1), positional),
+        version(1, TimeUnit.SECONDS.toMillis(2), positional)
+      ),
+      Option(Integer.valueOf(2))
+    )
+
+    encoded.map(_.importance) shouldBe List(true, true, false)
+    encoded.map(_.isCurrentlyPublic) shouldBe List(false, true, false)
   }
 }

From 0332da9cf4ea87dc560286107db811cb3d685851 Mon Sep 17 00:00:00 2001
From: Yang Zhang <yangz75@uci.edu>
Date: Sat, 22 Aug 2026 15:01:50 -0700
Subject: [PATCH 3/3] feat(frontend): choose between following and pinning in
 the share dialog

The share dialog said whether a workflow was public and nothing about
which copy the public was getting. With pinning that is now a choice, so
the dialog carries it: a two-option control -- Follow latest, or Pinned --
under the Public tile, plus a line naming the pinned version by the date
it went public.

Follow latest is the default and is exactly what publishing does today.
Pinning freezes the version the author has now; while a pin is in place
and their working copy has moved on, the panel says so and offers Update
to current, which is the only way those edits reach the public.

The panel re-reads its state when a save lands rather than on a timer:
`WorkflowPersistService` now emits on persist, rename and re-describe,
which is what makes "your edits are not public yet" appear as soon as the
autosave lands rather than the next time the dialog is opened. Publishing
returns the state it produced, so the panel does not have to ask again.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../user/workflow/WorkflowResource.scala      |   4 +
 common/config/src/main/resources/gui.conf     |   5 +
 .../texera/common/config/GuiConfig.scala      |   2 +
 .../texera/common/config/GuiConfigSpec.scala  |   5 +
 .../service/resource/ConfigResource.scala     |   1 +
 .../common/service/gui-config.service.mock.ts |   1 +
 .../workflow-persist.service.spec.ts          |  63 +++
 .../workflow-persist.service.ts               |  50 ++-
 frontend/src/app/common/type/gui-config.ts    |   1 +
 .../share-access/share-access.component.html  |  63 +++
 .../share-access/share-access.component.scss  | 156 ++++++-
 .../share-access.component.spec.ts            | 398 +++++++++++++++++-
 .../share-access/share-access.component.ts    | 150 ++++++-
 13 files changed, 861 insertions(+), 38 deletions(-)

diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
index e5d71281a67..8c464404f29 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala
@@ -735,6 +735,10 @@ class WorkflowResource extends LazyLogging {
     )
   }
 
+  /**
+    * Publishes the workflow. The public follows the author's latest content until a version is
+    * pinned; see [[pinLatest]].
+    */
   @PUT
   @RolesAllowed(Array("REGULAR", "ADMIN"))
   @Path("/public/{wid}")
diff --git a/common/config/src/main/resources/gui.conf b/common/config/src/main/resources/gui.conf
index 7cb7fff6555..6b6ef9994bc 100644
--- a/common/config/src/main/resources/gui.conf
+++ b/common/config/src/main/resources/gui.conf
@@ -95,6 +95,11 @@ gui {
     form-view-enabled = true
     form-view-enabled = ${?GUI_WORKFLOW_WORKSPACE_FORM_VIEW_ENABLED}
 
+    # whether an author may pin a version as the public copy of a published workflow, instead of
+    # the public always following their latest content
+    version-pinning-enabled = false
+    version-pinning-enabled = ${?GUI_WORKFLOW_WORKSPACE_VERSION_PINNING_ENABLED}
+
     # Whether to connect to local or production shared editing server. Set to true if you have
     # reverse proxy set up for y-websocket.
     production-shared-editing-server = false
diff --git a/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala
index d5295024dc8..26079eb240f 100644
--- a/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala
+++ b/common/config/src/main/scala/org/apache/texera/common/config/GuiConfig.scala
@@ -61,6 +61,8 @@ object GuiConfig {
     conf.getBoolean("gui.workflow-workspace.timetravel-enabled")
   val guiWorkflowWorkspaceFormViewEnabled: Boolean =
     conf.getBoolean("gui.workflow-workspace.form-view-enabled")
+  val guiWorkflowWorkspaceVersionPinningEnabled: Boolean =
+    conf.getBoolean("gui.workflow-workspace.version-pinning-enabled")
   val guiWorkflowWorkspaceProductionSharedEditingServer: Boolean =
     conf.getBoolean("gui.workflow-workspace.production-shared-editing-server")
   val guiWorkflowWorkspacePythonLanguageServerPort: String =
diff --git a/common/config/src/test/scala/org/apache/texera/common/config/GuiConfigSpec.scala b/common/config/src/test/scala/org/apache/texera/common/config/GuiConfigSpec.scala
index cfc978de3ab..5b7f7ca12c6 100644
--- a/common/config/src/test/scala/org/apache/texera/common/config/GuiConfigSpec.scala
+++ b/common/config/src/test/scala/org/apache/texera/common/config/GuiConfigSpec.scala
@@ -66,6 +66,11 @@ class GuiConfigSpec extends AnyFlatSpec with Matchers {
     ifUnset("GUI_WORKFLOW_WORKSPACE_FORM_VIEW_ENABLED")(
       GuiConfig.guiWorkflowWorkspaceFormViewEnabled shouldBe true
     )
+    // Version pinning stays off until the last PR of the series lands, so a half-built feature is
+    // never reachable from the dialog on a deployed instance.
+    ifUnset("GUI_WORKFLOW_WORKSPACE_VERSION_PINNING_ENABLED")(
+      GuiConfig.guiWorkflowWorkspaceVersionPinningEnabled shouldBe false
+    )
     ifUnset("GUI_WORKFLOW_WORKSPACE_PRODUCTION_SHARED_EDITING_SERVER")(
       GuiConfig.guiWorkflowWorkspaceProductionSharedEditingServer shouldBe false
     )
diff --git a/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala b/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala
index d682fe8d026..219b82b87c9 100644
--- a/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala
+++ b/config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala
@@ -92,6 +92,7 @@ class ConfigResource {
       "asyncRenderingEnabled" -> GuiConfig.guiWorkflowWorkspaceAsyncRenderingEnabled,
       "timetravelEnabled" -> GuiConfig.guiWorkflowWorkspaceTimetravelEnabled,
       "formViewEnabled" -> GuiConfig.guiWorkflowWorkspaceFormViewEnabled,
+      "versionPinningEnabled" -> GuiConfig.guiWorkflowWorkspaceVersionPinningEnabled,
       "productionSharedEditingServer" -> GuiConfig.guiWorkflowWorkspaceProductionSharedEditingServer,
       "defaultExecutionMode" -> GuiConfig.guiWorkflowWorkspaceDefaultExecutionMode,
       "workflowEmailNotificationEnabled" -> GuiConfig.guiWorkflowWorkspaceWorkflowEmailNotificationEnabled,
diff --git a/frontend/src/app/common/service/gui-config.service.mock.ts b/frontend/src/app/common/service/gui-config.service.mock.ts
index 8591fc727a9..2a3f43b02b2 100644
--- a/frontend/src/app/common/service/gui-config.service.mock.ts
+++ b/frontend/src/app/common/service/gui-config.service.mock.ts
@@ -43,6 +43,7 @@ export class MockGuiConfigService {
     asyncRenderingEnabled: false,
     timetravelEnabled: false,
     formViewEnabled: false,
+    versionPinningEnabled: false,
     productionSharedEditingServer: false,
     pythonLanguageServerPort: "3000",
     defaultDataTransferBatchSize: 100,
diff --git a/frontend/src/app/common/service/workflow-persist/workflow-persist.service.spec.ts b/frontend/src/app/common/service/workflow-persist/workflow-persist.service.spec.ts
index 7773458b5f1..c5246cb57f1 100644
--- a/frontend/src/app/common/service/workflow-persist/workflow-persist.service.spec.ts
+++ b/frontend/src/app/common/service/workflow-persist/workflow-persist.service.spec.ts
@@ -21,6 +21,7 @@ import { TestBed } from "@angular/core/testing";
 import { HttpClientTestingModule, HttpTestingController } from "@angular/common/http/testing";
 import {
   WorkflowPersistService,
+  WorkflowPublishStatus,
   WORKFLOW_BASE_URL,
   WORKFLOW_ID_URL,
   WORKFLOW_OWNER_URL,
@@ -99,6 +100,68 @@ describe("WorkflowPersistService", () => {
     httpTestingController.expectOne(request => request.method === "POST");
   });
 
+  it("should publish without pinning anything", () => {
+    service.updateWorkflowIsPublished(7, true).subscribe();
+    const request = httpTestingController.expectOne(req => req.url.endsWith("workflow/public/7"));
+    expect(request.request.method).toEqual("PUT");
+  });
+
+  it("should pin the latest version through the pin endpoint", () => {
+    let received: WorkflowPublishStatus | undefined;
+    service.pinLatestVersion(7).subscribe(status => (received = status));
+
+    const request = httpTestingController.expectOne(req => req.url.endsWith("workflow/pin/7"));
+    expect(request.request.method).toEqual("POST");
+    request.flush({
+      isPublished: true,
+      isPinned: true,
+      pinnedVersionTime: 1700000000000,
+      hasUnpublishedChanges: false,
+    });
+
+    expect(received?.pinnedVersionTime).toEqual(1700000000000);
+    expect(received?.hasUnpublishedChanges).toBe(false);
+  });
+
+  it("should drop the pin through the same path with DELETE", () => {
+    let received: WorkflowPublishStatus | undefined;
+    service.unpinVersion(7).subscribe(status => (received = status));
+
+    const request = httpTestingController.expectOne(req => req.url.endsWith("workflow/pin/7"));
+    expect(request.request.method).toEqual("DELETE");
+    request.flush({ isPublished: true, isPinned: false, hasUnpublishedChanges: false });
+
+    expect(received?.isPinned).toBe(false);
+  });
+
+  it("should announce a save once it lands, not when it is sent", () => {
+    // Panels that describe the saved copy re-read on this. Announcing on the request rather than the
+    // response would have them re-read the state the save was about to replace.
+    let announced = 0;
+    service.getWorkflowPersistedStream().subscribe(() => announced++);
+
+    const workflow = { wid: 7, name: "n", content: { operators: [], links: [] } } as unknown as Workflow;
+    service.persistWorkflow(workflow).subscribe();
+    const request = httpTestingController.expectOne(req => req.url.endsWith("workflow/persist"));
+    expect(announced).toBe(0);
+
+    request.flush({ wid: 7, name: "n", content: '{"operators":[]}' });
+
+    expect(announced).toBe(1);
+  });
+
+  it("should report unpublished changes from the publish status endpoint", () => {
+    let received: WorkflowPublishStatus | undefined;
+    service.getPublishStatus(7).subscribe(status => (received = status));
+
+    const request = httpTestingController.expectOne(req => req.url.endsWith("workflow/publish-status/7"));
+    expect(request.request.method).toEqual("GET");
+    request.flush({ isPublished: true, isPinned: true, hasUnpublishedChanges: true });
+
+    expect(received?.isPinned).toBe(true);
+    expect(received?.hasUnpublishedChanges).toBe(true);
+  });
+
   it("should check if workflow content and name returned correctly", () => {
     service
       .createWorkflow(jsonCast<WorkflowContent>(testContent), "testname")
diff --git a/frontend/src/app/common/service/workflow-persist/workflow-persist.service.ts b/frontend/src/app/common/service/workflow-persist/workflow-persist.service.ts
index fa61b112008..59e81ffb503 100644
--- a/frontend/src/app/common/service/workflow-persist/workflow-persist.service.ts
+++ b/frontend/src/app/common/service/workflow-persist/workflow-persist.service.ts
@@ -53,6 +53,17 @@ export const WORKFLOW_SET_DEFAULT_VIEW_URL = WORKFLOW_BASE_URL + "/set-default-v
 
 export const DEFAULT_WORKFLOW_NAME = "Untitled workflow";
 
+/** A published workflow follows the author's latest until they pin a version. */
+export interface WorkflowPublishStatus {
+  isPublished: boolean;
+  /** Whether a version is pinned. False means the public follows the author's latest. */
+  isPinned: boolean;
+  /** When the pinned version was created, which is how the dialog names it. */
+  pinnedVersionTime?: number;
+  /** Whether a pin is holding edits back. Always false while following. */
+  hasUnpublishedChanges: boolean;
+}
+
 @Injectable({
   providedIn: "root",
 })
@@ -85,6 +96,9 @@ export class WorkflowPersistService {
   private pendingSaves = 0;
   private readonly savesDrained = new Subject<void>();
 
+  /** Fires when a save lands: the write, not the keystroke, is what changes the saved copy. */
+  private workflowPersisted = new Subject<void>();
+
   constructor(
     private http: HttpClient,
     private notificationService: NotificationService,
@@ -172,7 +186,8 @@ export class WorkflowPersistService {
       })
       .pipe(
         filter((updatedWorkflow: Workflow) => updatedWorkflow != null),
-        map(WorkflowUtilService.parseWorkflowInfo)
+        map(WorkflowUtilService.parseWorkflowInfo),
+        tap(() => this.workflowPersisted.next())
       );
     // Replayed, so a caller that subscribes after the queue has already relayed the outcome (a
     // save that was quick, or a synchronous test double) still receives it.
@@ -182,6 +197,11 @@ export class WorkflowPersistService {
     return result.asObservable();
   }
 
+  /** Emits when a save lands, so panels describing the saved copy can re-read it. */
+  public getWorkflowPersistedStream(): Observable<void> {
+    return this.workflowPersisted.asObservable();
+  }
+
   /**
    * creates a workflow and insert it to backend database and return its information
    * @param newWorkflowName
@@ -274,6 +294,8 @@ export class WorkflowPersistService {
         name: name,
       })
       .pipe(
+        // A pin freezes the name too, so a rename is a save like any other.
+        tap(() => this.workflowPersisted.next()),
         catchError((error: unknown) => {
           // @ts-ignore
           this.notificationService.error(error.error.message);
@@ -292,6 +314,8 @@ export class WorkflowPersistService {
         description: description,
       })
       .pipe(
+        // Frozen by a pin like the name and the canvas are.
+        tap(() => this.workflowPersisted.next()),
         catchError((error: unknown) => {
           // @ts-ignore
           this.notificationService.error(error.error.message);
@@ -304,6 +328,10 @@ export class WorkflowPersistService {
     return this.http.get(`${AppSettings.getApiEndpoint()}/${WORKFLOW_BASE_URL}/type/${wid}`, { responseType: "text" });
   }
 
+  /**
+   * Publishes the workflow, or unpublishes it. A published workflow follows the author's latest
+   * content until a version is pinned; see {@link pinLatestVersion}.
+   */
   public updateWorkflowIsPublished(wid: number, isPublished: boolean): Observable<void> {
     if (isPublished) {
       return this.http.put<void>(`${AppSettings.getApiEndpoint()}/${WORKFLOW_BASE_URL}/public/${wid}`, null);
@@ -312,6 +340,26 @@ export class WorkflowPersistService {
     }
   }
 
+  /** Pins the author's current version as the public copy, so later edits stop reaching the public. */
+  public pinLatestVersion(wid: number): Observable<WorkflowPublishStatus> {
+    return this.http.post<WorkflowPublishStatus>(
+      `${AppSettings.getApiEndpoint()}/${WORKFLOW_BASE_URL}/pin/${wid}`,
+      null
+    );
+  }
+
+  /** Drops the pin, so the public follows the author's latest content again. */
+  public unpinVersion(wid: number): Observable<WorkflowPublishStatus> {
+    return this.http.delete<WorkflowPublishStatus>(`${AppSettings.getApiEndpoint()}/${WORKFLOW_BASE_URL}/pin/${wid}`);
+  }
+
+  /** Whether the workflow is published, whether a version is pinned, and whether it holds edits back. */
+  public getPublishStatus(wid: number): Observable<WorkflowPublishStatus> {
+    return this.http.get<WorkflowPublishStatus>(
+      `${AppSettings.getApiEndpoint()}/${WORKFLOW_BASE_URL}/publish-status/${wid}`
+    );
+  }
+
   public setWorkflowPersistFlag(flag: boolean): void {
     this.workflowPersistFlag = flag;
   }
diff --git a/frontend/src/app/common/type/gui-config.ts b/frontend/src/app/common/type/gui-config.ts
index 5b8b23db03d..d24c4805bfb 100644
--- a/frontend/src/app/common/type/gui-config.ts
+++ b/frontend/src/app/common/type/gui-config.ts
@@ -34,6 +34,7 @@ export interface GuiConfig {
   asyncRenderingEnabled: boolean;
   timetravelEnabled: boolean;
   formViewEnabled: boolean;
+  versionPinningEnabled: boolean;
   productionSharedEditingServer: boolean;
   pythonLanguageServerPort: string;
   defaultDataTransferBatchSize: number;
diff --git a/frontend/src/app/dashboard/component/user/share-access/share-access.component.html b/frontend/src/app/dashboard/component/user/share-access/share-access.component.html
index 8128f333633..bc77db1cea3 100644
--- a/frontend/src/app/dashboard/component/user/share-access/share-access.component.html
+++ b/frontend/src/app/dashboard/component/user/share-access/share-access.component.html
@@ -55,6 +55,69 @@
   </button>
 </div>
 
+<!--
+  A public workflow either follows the author's latest content or shows the version the author
+  pinned. Two states, so the control is a two-way switch and everything under it describes the side
+  it is on -- a fact, not a warning: keeping an older version public is a legitimate choice. Only
+  shown while Public is selected, and only to someone who can act on it.
+-->
+<div
+  class="publish-anchor"
+  [attr.data-state]="publishState"
+  *ngIf="config.env.versionPinningEnabled && publishStatus && isPublic && type === 'workflow'">
+  <!--
+    A switch rather than a button labelled with the other state: both choices stay on screen, so the
+    author reads what the options are instead of inferring one from the label of the other.
+  -->
+  <nz-segmented
+    class="publish-seg"
+    nzBlock
+    [nzOptions]="publicCopyOptions"
+    [nzDisabled]="!hasWriteAccess || isPinning"
+    [ngModel]="publishStatus.isPinned ? 1 : 0"
+    (nzValueChange)="choosePublicCopy($event === 1)"></nz-segmented>
+
+  <!-- One sentence for the side the switch is on, with the part that answers "what do they see" in bold. -->
+  <div class="publish-status">
+    <!-- Green while the public has everything, amber once a pin is holding a version back. -->
+    <nz-badge [nzStatus]="publishState === 'follow' ? 'success' : 'warning'"></nz-badge>
+    <span
+      class="publish-say"
+      [ngSwitch]="publishState">
+      <ng-container *ngSwitchCase="'behind'">The public is on an <b>older version</b>.</ng-container>
+      <ng-container *ngSwitchCase="'pinned'">The public sees <b>your current version</b>, pinned.</ng-container>
+      <ng-container *ngSwitchDefault>The public sees <b>your latest</b>, updated on every save.</ng-container>
+    </span>
+  </div>
+
+  <!--
+    Only the pinned side has more to say. Behind, that is the one decision the state leaves open, so
+    it gets a card: which version is out there, what it costs, and the act that ends it.
+  -->
+  <ng-container *ngIf="publishStatus.isPinned">
+    <div
+      class="publish-card"
+      *ngIf="publishStatus.hasUnpublishedChanges; else pinnedInSync">
+      <div class="publish-card-text">
+        <div class="publish-card-title">
+          Pinned to {{ publishStatus.pinnedVersionTime | date: publicationTimeFormat }}
+        </div>
+        <div class="publish-card-meta">Your later edits stay private</div>
+      </div>
+      <button
+        type="button"
+        class="publish-card-action"
+        [disabled]="!hasWriteAccess || isPinning"
+        (click)="choosePublicCopy(true, true)">
+        Update to current
+      </button>
+    </div>
+    <ng-template #pinnedInSync>
+      <div class="publish-hint">Frozen as you keep editing. Restore it any time from the version panel.</div>
+    </ng-template>
+  </ng-container>
+</div>
+
 <form
   [formGroup]="validateForm"
   (ngSubmit)="grantAccess()">
diff --git a/frontend/src/app/dashboard/component/user/share-access/share-access.component.scss b/frontend/src/app/dashboard/component/user/share-access/share-access.component.scss
index 1a74945e159..401747d706b 100644
--- a/frontend/src/app/dashboard/component/user/share-access/share-access.component.scss
+++ b/frontend/src/app/dashboard/component/user/share-access/share-access.component.scss
@@ -36,25 +36,163 @@
   line-height: 1;
 }
 
-.access-button-group {
-  display: flex;
-  flex-direction: row;
-  justify-content: space-evenly;
-  align-items: center;
-}
-
 .access-button {
   display: flex;
+  flex: 1;
   flex-direction: column;
   align-items: center;
   justify-content: space-evenly;
   height: 150px;
-  width: 300px;
   padding: 20px;
-  margin-bottom: 24px;
   border-radius: 10px;
 }
 
+// Laid out with a gap rather than space-evenly so the tiles span the full width of the dialog body.
+// space-evenly left them inset, which the full-width publish-state strip below could not line up with.
+.access-button-group {
+  display: flex;
+  gap: 24px;
+  align-items: stretch;
+  margin-bottom: 24px;
+}
+
+.access-button-group:has(+ .publish-anchor) {
+  margin-bottom: 8px;
+}
+
+// The publish panel. Nested under one root so nothing here reaches the Private/Public tiles, the
+// invite form or the access list around it.
+.publish-anchor {
+  margin-bottom: 24px;
+
+  // Sized to the panel's own scale; everything else about the switch is the component's.
+  // ::ng-deep because these are the control's own nodes: our styles are scoped to this component,
+  // and its inner DOM belongs to ng-zorro.
+  .publish-seg ::ng-deep {
+    font-size: 13.5px;
+    // Between antd's default and a pill: square enough to sit with the tiles above, round enough to
+    // echo the card below, whose corners are 12px.
+    border-radius: 12px;
+
+    .ant-segmented-item,
+    .ant-segmented-thumb {
+      border-radius: 9px;
+    }
+
+    .ant-segmented-item-label {
+      padding: 3px 16px;
+    }
+
+    // Quicker than the default: the thumb only confirms a choice the author already made, and a
+    // slide they have to wait out reads as the app thinking rather than as an answer.
+    .ant-segmented-thumb {
+      transition-duration: 0.16s;
+    }
+
+    // The side in force takes the colour of the state it puts the workflow in -- green while the
+    // public follows along, amber once a version is held. Without it the only mark is a white tile,
+    // which is easy to miss at a glance.
+    .ant-segmented-item-label[aria-selected="true"] {
+      font-weight: 600;
+      color: #12a05e;
+    }
+  }
+
+  // Amber once a version is held, matching the dot and the sentence below.
+  &:not([data-state="follow"]) .publish-seg ::ng-deep .ant-segmented-item-label[aria-selected="true"] {
+    color: #e0951c;
+  }
+
+  .publish-status {
+    display: flex;
+    align-items: center;
+    gap: 12px;
+    padding-top: 16px;
+  }
+
+  .publish-say {
+    flex: 1;
+    min-width: 0;
+    font-size: 14px;
+    line-height: 1.45;
+    color: #5f666f;
+
+    b {
+      font-weight: 580;
+      color: #1c1f26;
+    }
+  }
+
+  // The one state that leaves the author something to decide gets a card: which version is public,
+  // what it costs them, and the act that ends it -- in that order, left to right.
+  .publish-card {
+    display: flex;
+    align-items: center;
+    gap: 12px;
+    margin-top: 12px;
+    padding: 12px 12px 12px 16px;
+    background: #f6f7f9;
+    border: 1px solid #e6e9ee;
+    border-radius: 12px;
+  }
+
+  .publish-card-text {
+    flex: 1;
+    min-width: 0;
+  }
+
+  .publish-card-title {
+    font-size: 13px;
+    font-weight: 580;
+    color: #1c1f26;
+    // Digits of equal width, so a date that ticks does not shuffle the words around it.
+    font-variant-numeric: tabular-nums;
+  }
+
+  .publish-card-meta {
+    margin-top: 2px;
+    font-size: 11.5px;
+    line-height: 1.3;
+    color: #9299a2;
+  }
+
+  .publish-card-action {
+    flex: none;
+    padding: 9px 14px;
+    border: none;
+    border-radius: 8px;
+    background: #ddeafe;
+    color: #2f74f0;
+    font: inherit;
+    font-size: 13px;
+    font-weight: 580;
+    white-space: nowrap;
+    cursor: pointer;
+    transition: 0.14s;
+
+    &:hover:not([disabled]) {
+      background: #d0e1fd;
+    }
+
+    &[disabled] {
+      cursor: default;
+      opacity: 0.55;
+    }
+  }
+
+  .publish-hint {
+    margin-top: 12px;
+    font-size: 12.5px;
+    line-height: 1.5;
+    color: #9299a2;
+
+    b {
+      font-weight: 540;
+      color: #5f666f;
+    }
+  }
+}
+
 .button-icon {
   font-size: 40px;
 }
diff --git a/frontend/src/app/dashboard/component/user/share-access/share-access.component.spec.ts b/frontend/src/app/dashboard/component/user/share-access/share-access.component.spec.ts
index ec41db898c2..c1ee977df05 100644
--- a/frontend/src/app/dashboard/component/user/share-access/share-access.component.spec.ts
+++ b/frontend/src/app/dashboard/component/user/share-access/share-access.component.spec.ts
@@ -22,7 +22,7 @@ import { HttpClientTestingModule } from "@angular/common/http/testing";
 import { NoopAnimationsModule } from "@angular/platform-browser/animations";
 import { By } from "@angular/platform-browser";
 import { HttpErrorResponse } from "@angular/common/http";
-import { of, throwError } from "rxjs";
+import { Subject, of, throwError } from "rxjs";
 
 import { NZ_MODAL_DATA, NzModalRef, NzModalService } from "ng-zorro-antd/modal";
 import { NzMessageService } from "ng-zorro-antd/message";
@@ -35,6 +35,7 @@ import { NotificationService } from "../../../../common/service/notification/not
 import { DatasetService } from "../../../service/user/dataset/dataset.service";
 import { WorkflowPersistService } from "src/app/common/service/workflow-persist/workflow-persist.service";
 import { WorkflowActionService } from "src/app/workspace/service/workflow-graph/model/workflow-action.service";
+import { GuiConfigService } from "src/app/common/service/gui-config.service";
 import { Privilege } from "../../../type/share-access.interface";
 import { ResourceRegistryService } from "../../../service/user/resource-registry/resource-registry.service";
 import { ResourceDescriptor } from "../../../type/resource-descriptor";
@@ -47,6 +48,7 @@ interface SetupOptions {
   currentEmail?: string | undefined;
   /** What the registry answers for the kind, in place of the shipped descriptor. */
   descriptor?: ResourceDescriptor;
+  versionPinningEnabled?: boolean;
 }
 
 /**
@@ -71,7 +73,7 @@ describe("ShareAccessComponent", () => {
   let messageSpy: { error: ReturnType<typeof vi.fn> };
   let modalRefSpy: { close: ReturnType<typeof vi.fn> };
   let modalServiceSpy: { create: ReturnType<typeof vi.fn> };
-  let workflowPersistSpy: {
+  let workflowPersistSpy: any /* the publish-status spies vary per test */ & {
     getWorkflowIsPublished: ReturnType<typeof vi.fn>;
     updateWorkflowIsPublished: ReturnType<typeof vi.fn>;
   };
@@ -79,7 +81,10 @@ describe("ShareAccessComponent", () => {
     getDataset: ReturnType<typeof vi.fn>;
     updateDatasetPublicity: ReturnType<typeof vi.fn>;
   };
-  let workflowActionSpy: { setWorkflowIsPublished: ReturnType<typeof vi.fn> };
+  let workflowActionSpy: {
+    setWorkflowIsPublished: ReturnType<typeof vi.fn>;
+    getWorkflow: ReturnType<typeof vi.fn>;
+  };
   let userServiceCurrentEmail: string | undefined;
   // The component reads publicity back after writing it, so the doubles have to hold state:
   // the workflow endpoint sets it absolutely, the dataset one toggles.
@@ -87,6 +92,8 @@ describe("ShareAccessComponent", () => {
   let datasetPublished: boolean;
   let capturedModalConfigs: any[];
   /** The NzModalRef stubs handed back by modalService.create, in creation order. */
+  /** Stands in for the editor's autosave landing, which is what the panel has to follow. */
+  let persisted: Subject<any>;
   let capturedModalRefs: { close: ReturnType<typeof vi.fn> }[];
   /**
    * The fixture built by the most recent setupComponent() call, for the template-level tests.
@@ -96,7 +103,15 @@ describe("ShareAccessComponent", () => {
   let fixture: ComponentFixture<ShareAccessComponent>;
 
   function setupComponent(opts: SetupOptions = {}): ShareAccessComponent {
-    const { type = "workflow", id = 1, inWorkspace = false, currentEmail = "me@example.com" } = opts;
+    const {
+      type = "workflow",
+      id = 1,
+      inWorkspace = false,
+      currentEmail = "me@example.com",
+      // On for the cases that are about the panel; the feature ships behind this flag, and the
+      // case below covers what the dialog looks like while it is off.
+      versionPinningEnabled = true,
+    } = opts;
     userServiceCurrentEmail = currentEmail;
 
     TestBed.configureTestingModule({
@@ -118,6 +133,7 @@ describe("ShareAccessComponent", () => {
         { provide: WorkflowPersistService, useValue: workflowPersistSpy },
         { provide: DatasetService, useValue: datasetServiceSpy },
         { provide: WorkflowActionService, useValue: workflowActionSpy },
+        { provide: GuiConfigService, useValue: { env: { versionPinningEnabled } } },
       ],
     });
     if (opts.descriptor) {
@@ -128,9 +144,34 @@ describe("ShareAccessComponent", () => {
     return fixture.componentInstance;
   }
 
+  /** Everything the publish panel renders, for the assertions about what the author actually reads. */
+  function publishLineText(): string {
+    fixture.detectChanges();
+    return (fixture.nativeElement.querySelector(".publish-anchor")?.textContent ?? "").replace(/\s+/g, " ").trim();
+  }
+
+  /**
+   * The two sides of the switch. Which one is highlighted is the control's own business -- it paints
+   * the selection from an animation frame, which this environment never runs -- so these assert the
+   * labels and leave the state itself to `publishState`.
+   */
+  function segments(): string[] {
+    fixture.detectChanges();
+    return Array.from(fixture.nativeElement.querySelectorAll(".ant-segmented-item-label")).map((b: any) =>
+      (b.textContent ?? "").trim()
+    );
+  }
+
+  /** The card under the line, which only a pin holding edits back puts on screen. */
+  function publishNoteText(): string {
+    fixture.detectChanges();
+    return (fixture.nativeElement.querySelector(".publish-card")?.textContent ?? "").replace(/\s+/g, " ").trim();
+  }
+
   beforeEach(() => {
     TestBed.resetTestingModule();
     fixture = undefined as unknown as ComponentFixture<ShareAccessComponent>;
+    persisted = new Subject<any>();
     capturedModalConfigs = [];
     capturedModalRefs = [];
     gmailSpy = { sendEmail: vi.fn() };
@@ -153,12 +194,19 @@ describe("ShareAccessComponent", () => {
     };
     workflowPublished = false;
     datasetPublished = false;
+    const followingStatus = { isPublished: true, isPinned: false, hasUnpublishedChanges: false };
     workflowPersistSpy = {
       getWorkflowIsPublished: vi.fn(() => of(workflowPublished ? "Public" : "Private")),
       updateWorkflowIsPublished: vi.fn((_id: number, next: boolean) => {
         workflowPublished = next;
         return of(null);
       }),
+      getPublishStatus: vi.fn().mockReturnValue(of(followingStatus)),
+      getWorkflowPersistedStream: vi.fn().mockReturnValue(persisted.asObservable()),
+      isWorkflowPersistEnabled: vi.fn().mockReturnValue(true),
+      persistWorkflow: vi.fn().mockReturnValue(of({ wid: 3 })),
+      pinLatestVersion: vi.fn().mockReturnValue(of({ ...followingStatus, isPinned: true })),
+      unpinVersion: vi.fn().mockReturnValue(of(followingStatus)),
     };
     datasetServiceSpy = {
       getDataset: vi.fn(() => of({ dataset: { isPublic: datasetPublished } })),
@@ -167,7 +215,11 @@ describe("ShareAccessComponent", () => {
         return of(null);
       }),
     };
-    workflowActionSpy = { setWorkflowIsPublished: vi.fn() };
+    workflowActionSpy = {
+      setWorkflowIsPublished: vi.fn(),
+      // The publish panel never writes; it only re-reads once a save of the author's own lands.
+      getWorkflow: vi.fn().mockReturnValue({ wid: 3, name: "w", content: { operators: [], links: [] } }),
+    };
   });
 
   function getFooterButton(config: any, label: string): { onClick: () => void } {
@@ -1127,4 +1179,340 @@ describe("ShareAccessComponent", () => {
       expect(datasetServiceSpy.updateDatasetPublicity).not.toHaveBeenCalled();
     });
   });
+
+  describe("publish state line", () => {
+    // The owner of the workflow, so hasWriteAccess is true.
+    const asOwner = { currentEmail: "owner@example.com" };
+
+    it("does not fetch publish status for an unpublished workflow", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Private"));
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(workflowPersistSpy.getPublishStatus).not.toHaveBeenCalled();
+      expect(c.publishStatus).toBeUndefined();
+    });
+
+    it("reports unpublished changes on a pinned workflow", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(workflowPersistSpy.getPublishStatus).toHaveBeenCalledWith(3);
+      expect(c.publishStatus?.hasUnpublishedChanges).toBe(true);
+    });
+
+    it("does not fetch publish status for a viewer who cannot publish", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      accessServiceSpy.getAccessList.mockReturnValue(
+        of([{ email: "reader@example.com", name: "R", privilege: Privilege.READ }])
+      );
+      const c = setupComponent({ type: "workflow", id: 3, currentEmail: "reader@example.com" });
+      expect(workflowPersistSpy.getPublishStatus).not.toHaveBeenCalled();
+      expect(c.publishStatus).toBeUndefined();
+    });
+
+    it("clears the pending state after moving the pin forward", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      workflowPersistSpy.pinLatestVersion.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      c.choosePublicCopy(true, true);
+
+      expect(workflowPersistSpy.pinLatestVersion).toHaveBeenCalledWith(3);
+      expect(c.publishStatus?.hasUnpublishedChanges).toBe(false);
+      expect(c.isPinning).toBe(false);
+    });
+
+    it("goes back to following when the pin is dropped", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      workflowPersistSpy.unpinVersion.mockReturnValue(
+        of({ isPublished: true, isPinned: false, hasUnpublishedChanges: false })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      c.choosePublicCopy(false);
+
+      expect(workflowPersistSpy.unpinVersion).toHaveBeenCalledWith(3);
+      expect(c.publishStatus?.isPinned).toBe(false);
+      // Nothing is held back once the public follows the latest.
+      expect(c.publishStatus?.hasUnpublishedChanges).toBe(false);
+      expect(c.isPinning).toBe(false);
+    });
+
+    it("names the pinned version by its date once the author has moved on", () => {
+      // The date is how the same version is named in the revision panel, which is where the author
+      // goes to restore it. Naming it any other way would leave them matching two descriptions.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({
+          isPublished: true,
+          isPinned: true,
+          hasUnpublishedChanges: true,
+          pinnedVersionTime: Date.parse("2026-08-12T23:59:00"),
+        })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishLineText()).toContain("Pinned to Aug 12, 23:59:00");
+      expect(publishLineText()).toContain("Your later edits stay private");
+    });
+
+    it("says the pinned copy is the current one rather than dating it", () => {
+      // A date here would invite the author to work out whether it is still what they have; saying
+      // so is the answer to the question they would be asking.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({
+          isPublished: true,
+          isPinned: true,
+          hasUnpublishedChanges: false,
+          pinnedVersionTime: Date.parse("2026-08-12T23:59:00"),
+        })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishLineText()).toContain("your current version");
+      expect(publishLineText()).not.toContain("Aug 12");
+    });
+
+    it("prints the pinned time to the second, as the revision panel does", () => {
+      // The line names the version the revision panel marks, so the two must print it identically --
+      // one format, never varied, is what guarantees that.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({
+          isPublished: true,
+          isPinned: true,
+          hasUnpublishedChanges: true,
+          pinnedVersionTime: Date.parse("2026-08-12T23:58:51"),
+        })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishLineText()).toContain("Aug 12, 23:58:51");
+    });
+
+    it("says the same thing about both states, with only the answer changing", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: false, hasUnpublishedChanges: false })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishLineText()).toContain("The public sees your latest");
+      expect(segments()).toEqual(["Follow latest", "Pinned"]);
+    });
+
+    it("drops the line when the workflow is unpublished", () => {
+      workflowPublished = true;
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(c.publishStatus).toBeDefined();
+
+      c.setPublished(false);
+
+      expect(c.isPublic).toBe(false);
+      expect(c.publishStatus).toBeUndefined();
+    });
+
+    it("hides the notice when the status request fails", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(throwError(() => new Error("boom")));
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(c.publishStatus).toBeUndefined();
+    });
+
+    it("surfaces a failed pin and stops the spinner", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      workflowPersistSpy.pinLatestVersion.mockReturnValue(
+        throwError(() => new HttpErrorResponse({ error: { message: "nope" }, status: 500 }))
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      c.choosePublicCopy(true, true);
+
+      expect(notificationSpy.error).toHaveBeenCalledWith("nope");
+      expect(c.isPinning).toBe(false);
+      // The pending state stands, so the author can try again.
+      expect(c.publishStatus?.hasUnpublishedChanges).toBe(true);
+    });
+
+    it("never asks for publish status on a dataset", () => {
+      const c = setupComponent({ type: "dataset", id: 12, ...asOwner });
+      expect(workflowPersistSpy.getPublishStatus).not.toHaveBeenCalled();
+      expect(c.publishStatus).toBeUndefined();
+    });
+
+    it("fetches the status after publishing from private", () => {
+      workflowPublished = false;
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(workflowPersistSpy.getPublishStatus).not.toHaveBeenCalled();
+
+      c.setPublished(true);
+
+      expect(workflowPersistSpy.updateWorkflowIsPublished).toHaveBeenCalledWith(3, true);
+      expect(workflowPersistSpy.getPublishStatus).toHaveBeenCalledWith(3);
+      expect(c.publishStatus).toBeDefined();
+    });
+
+    it("offers both choices, with the state itself naming the one in force", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(segments()).toEqual(["Follow latest", "Pinned"]);
+      expect(c.publishState).toBe("pinned");
+    });
+
+    it("does nothing when the side already in force is picked again", () => {
+      // Re-picking "Pinned" would otherwise publish whatever has been edited since -- which is the
+      // card's job, and the card says what it would publish.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      c.choosePublicCopy(true);
+
+      expect(workflowPersistSpy.pinLatestVersion).not.toHaveBeenCalled();
+      expect(c.publishStatus?.hasUnpublishedChanges).toBe(true);
+    });
+
+    it("does nothing when following is picked while already following", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: false, hasUnpublishedChanges: false })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      c.choosePublicCopy(false);
+
+      expect(workflowPersistSpy.unpinVersion).not.toHaveBeenCalled();
+    });
+
+    it("names the pinned version and the way out while it is holding edits back", () => {
+      // The card is the answer to "my edits are not public": which version is out there, what it
+      // costs, and the act that ends it. In every other state there is nothing to decide.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({
+          isPublished: true,
+          isPinned: true,
+          hasUnpublishedChanges: true,
+          pinnedVersionTime: Date.parse("2026-08-12T23:59:00"),
+        })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishNoteText()).toContain("Pinned to Aug 12, 23:59:00");
+      expect(publishNoteText()).toContain("Your later edits stay private");
+      expect(publishNoteText()).toContain("Update to current");
+    });
+
+    it("points at the version panel once the pinned copy is what the author has", () => {
+      // The recovery path when they do move on, named where they are already looking.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishLineText()).toContain("Restore it any time from the version panel");
+    });
+
+    it("shows nothing and asks nothing while the feature is off", () => {
+      // The series ships dark: until the last PR turns the flag on, the dialog is exactly what it
+      // is today, and it does not even ask the server about a state it cannot show.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+
+      setupComponent({ type: "workflow", id: 3, versionPinningEnabled: false, ...asOwner });
+
+      expect(publishLineText()).toBe("");
+      expect(workflowPersistSpy.getPublishStatus).not.toHaveBeenCalled();
+    });
+
+    it("writes nothing when it opens", () => {
+      // The dialog reports on the saved copy rather than forcing the canvas to become it: the canvas
+      // is empty while a workflow loads, and a save nobody asked for could write that emptiness over
+      // every operator the workflow had.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      setupComponent({ type: "workflow", id: 3, inWorkspace: true, ...asOwner });
+
+      expect(workflowPersistSpy.persistWorkflow).not.toHaveBeenCalled();
+    });
+
+    it("re-reads the state when a save lands", () => {
+      // The editor saves on a debounce, so the answer the dialog fetched when it opened describes a
+      // workflow the author may already have moved past. Nothing else tells it -- the canvas changing
+      // is not the save landing, and until the save lands the server still answers with the old copy.
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(publishNoteText()).toBe("");
+
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      persisted.next({ wid: 3 });
+
+      expect(workflowPersistSpy.getPublishStatus).toHaveBeenCalledTimes(2);
+      expect(publishNoteText()).toContain("Update to current");
+    });
+
+    it("says nothing extra when the pinned copy is what the author has", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      expect(publishNoteText()).toBe("");
+    });
+
+    it("publishes the edits a pin was holding back, from the note", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Public"));
+      workflowPersistSpy.getPublishStatus.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: true })
+      );
+      workflowPersistSpy.pinLatestVersion.mockReturnValue(
+        of({ isPublished: true, isPinned: true, hasUnpublishedChanges: false })
+      );
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+
+      fixture.nativeElement.querySelector(".publish-card-action").click();
+
+      expect(workflowPersistSpy.pinLatestVersion).toHaveBeenCalledWith(3);
+      expect(c.publishStatus?.hasUnpublishedChanges).toBe(false);
+      // Answered, so the question goes away.
+      expect(publishNoteText()).toBe("");
+    });
+
+    it("says nothing about the copies before the status arrives", () => {
+      workflowPersistSpy.getWorkflowIsPublished.mockReturnValue(of("Private"));
+      const c = setupComponent({ type: "workflow", id: 3, ...asOwner });
+      expect(c.publishStatus).toBeUndefined();
+      expect(publishLineText()).toBe("");
+    });
+  });
 });
diff --git a/frontend/src/app/dashboard/component/user/share-access/share-access.component.ts b/frontend/src/app/dashboard/component/user/share-access/share-access.component.ts
index b5baf5d9dba..fe631351fed 100644
--- a/frontend/src/app/dashboard/component/user/share-access/share-access.component.ts
+++ b/frontend/src/app/dashboard/component/user/share-access/share-access.component.ts
@@ -27,13 +27,18 @@ import { GmailService } from "../../../../common/service/gmail/gmail.service";
 import { NZ_MODAL_DATA, NzModalRef, NzModalService } from "ng-zorro-antd/modal";
 import { NotificationService } from "../../../../common/service/notification/notification.service";
 import { HttpErrorResponse } from "@angular/common/http";
-import { catchError, of, switchMap } from "rxjs";
+import { catchError, forkJoin, of, switchMap } from "rxjs";
 import { NzMessageService } from "ng-zorro-antd/message";
 import { WorkflowActionService } from "src/app/workspace/service/workflow-graph/model/workflow-action.service";
+import {
+  WorkflowPersistService,
+  WorkflowPublishStatus,
+} from "src/app/common/service/workflow-persist/workflow-persist.service";
+import { GuiConfigService } from "../../../../common/service/gui-config.service";
 import { ResourceRegistryService } from "../../../service/user/resource-registry/resource-registry.service";
 import { ResourceDescriptor } from "../../../type/resource-descriptor";
 import { EntityType } from "../../../../hub/service/hub.service";
-import { NgIf, NgFor } from "@angular/common";
+import { NgIf, NgFor, NgSwitch, NgSwitchCase, NgSwitchDefault, DatePipe } from "@angular/common";
 import { NzSpaceCompactItemDirective } from "ng-zorro-antd/space";
 import { NzButtonComponent } from "ng-zorro-antd/button";
 import { NzWaveDirective } from "ng-zorro-antd/core/wave";
@@ -46,6 +51,8 @@ import { NzInputDirective } from "ng-zorro-antd/input";
 import { NzAutocompleteTriggerDirective, NzAutocompleteComponent } from "ng-zorro-antd/auto-complete";
 import { NzTagComponent } from "ng-zorro-antd/tag";
 import { NzTooltipDirective } from "ng-zorro-antd/tooltip";
+import { NzSegmentedComponent } from "ng-zorro-antd/segmented";
+import { NzBadgeComponent } from "ng-zorro-antd/badge";
 
 @UntilDestroy()
 @Component({
@@ -73,6 +80,12 @@ import { NzTooltipDirective } from "ng-zorro-antd/tooltip";
     NgFor,
     NzTagComponent,
     NzTooltipDirective,
+    NzSegmentedComponent,
+    NzBadgeComponent,
+    NgSwitch,
+    NgSwitchCase,
+    NgSwitchDefault,
+    DatePipe,
   ],
 })
 export class ShareAccessComponent implements OnInit, OnDestroy {
@@ -91,6 +104,16 @@ export class ShareAccessComponent implements OnInit, OnDestroy {
   isPublic: boolean | null = null;
   /** Undefined for kinds the registry does not carry, i.e. computing units. */
   private readonly descriptor: ResourceDescriptor | undefined;
+  /** Undefined until fetched, and for anyone who cannot publish (the endpoint needs write access). */
+  publishStatus?: WorkflowPublishStatus;
+  isPinning = false;
+  /** The two sides of the switch; the control only takes strings or numbers, so 1 means pinned. */
+  readonly publicCopyOptions = [
+    { label: "Follow latest", value: 0 },
+    { label: "Pinned", value: 1 },
+  ];
+  /** To the second, always: the revision panel names the same version that way. */
+  readonly publicationTimeFormat = "MMM d, HH:mm:ss";
   private shouldRefresh = false;
   @Output() refresh = new EventEmitter<void>();
 
@@ -103,7 +126,9 @@ export class ShareAccessComponent implements OnInit, OnDestroy {
     private message: NzMessageService,
     private modalService: NzModalService,
     private workflowActionService: WorkflowActionService,
+    private workflowPersistService: WorkflowPersistService,
     private resourceRegistry: ResourceRegistryService,
+    protected config: GuiConfigService,
     private modalRef: NzModalRef
   ) {
     this.validateForm = this.formBuilder.group({
@@ -114,6 +139,61 @@ export class ShareAccessComponent implements OnInit, OnDestroy {
     this.descriptor = this.resourceRegistry.find(this.type as EntityType);
   }
 
+  /** Only for a published workflow the user can publish -- the endpoint requires write access. */
+  private refreshPublishStatus(): void {
+    if (!this.config.env.versionPinningEnabled) {
+      return;
+    }
+    if (this.type !== "workflow" || !this.isPublic || !this.hasWriteAccess) {
+      this.publishStatus = undefined;
+      return;
+    }
+    this.workflowPersistService
+      .getPublishStatus(this.id)
+      .pipe(untilDestroyed(this))
+      .subscribe({
+        next: status => (this.publishStatus = status),
+        error: () => (this.publishStatus = undefined),
+      });
+  }
+
+  /** Named as a state, so the switch, the sentence and the dot cannot disagree about which is in force. */
+  get publishState(): "follow" | "pinned" | "behind" {
+    if (!this.publishStatus?.isPinned) {
+      return "follow";
+    }
+    return this.publishStatus.hasUnpublishedChanges ? "behind" : "pinned";
+  }
+
+  /**
+   * Changes what the public sees, leaving the author's working copy alone. Picking the side already
+   * in force does nothing, so a stray click cannot silently publish edits made since the pin;
+   * `republish` is how the card asks for exactly that.
+   */
+  public choosePublicCopy(pinned: boolean, republish = false): void {
+    if (!this.publishStatus || (pinned === this.publishStatus.isPinned && !republish)) {
+      return;
+    }
+    this.isPinning = true;
+    const change = pinned
+      ? { request: this.workflowPersistService.pinLatestVersion(this.id), done: "The public now sees this version" }
+      : { request: this.workflowPersistService.unpinVersion(this.id), done: "The public now follows your latest" };
+    change.request
+      .pipe(untilDestroyed(this))
+      .subscribe({
+        next: status => {
+          this.publishStatus = status;
+          this.notificationService.success(change.done);
+        },
+        error: (error: unknown) => {
+          if (error instanceof HttpErrorResponse) {
+            this.notificationService.error(error.error.message);
+          }
+        },
+      })
+      .add(() => (this.isPinning = false));
+  }
+
   get hasWriteAccess(): boolean {
     if (!this.currentEmail) {
       return false;
@@ -126,35 +206,54 @@ export class ShareAccessComponent implements OnInit, OnDestroy {
   }
 
   ngOnInit(): void {
-    this.accessService
-      .getAccessList(this.type, this.id)
-      .pipe(untilDestroyed(this))
-      .subscribe(access => (this.accessList = access));
-    this.accessService
-      .getOwner(this.type, this.id)
-      .pipe(untilDestroyed(this))
-      .subscribe(name => {
-        this.owner = name;
-      });
-    // Stays null for kinds that cannot be published, which is what hides the publish buttons.
-    // A failed request leaves it null too, and the buttons are equally gone, so say so: without
-    // this the dialog looked complete while quietly offering one control fewer, and the only way
-    // to find out was the network tab.
-    this.descriptor
-      ?.isPublic?.(this.id)
+    // Joined rather than subscribed separately because the publish state depends on all three:
+    // hasWriteAccess is only answerable once the owner and the access list have landed, and the
+    // panel only applies to a published workflow.
+    // Stays undefined for kinds that cannot be published, which is what hides the publish buttons.
+    forkJoin([
+      this.accessService.getAccessList(this.type, this.id),
+      this.accessService.getOwner(this.type, this.id),
+      this.descriptor?.isPublic?.(this.id) ?? of(undefined),
+    ])
       .pipe(untilDestroyed(this))
       .subscribe({
-        next: isPublic => (this.isPublic = isPublic),
+        next: ([accessList, owner, isPublic]) => {
+          this.accessList = accessList;
+          this.owner = owner;
+          if (isPublic !== undefined) {
+            this.isPublic = isPublic;
+          }
+          if (this.type !== "workflow") {
+            return;
+          }
+          // Answered from the saved copy, which the editor writes a few seconds after the last edit.
+          // Deliberately not forced to save first: the canvas is not always the workflow -- it is
+          // empty while one loads, and stays empty if the collaborative model never arrives -- so a
+          // save nobody asked for could write that emptiness over every operator the workflow had.
+          // The subscription below picks the answer up again as soon as the autosave lands.
+          this.refreshPublishStatus();
+        },
         error: () => {
-          // `ngOnInit` is re-entered as a refresh after an access change, so a value from the
-          // previous read may still be here. Drop it: keeping it would leave the buttons on
-          // screen, showing a state nothing has confirmed, while the toast says they are gone.
+          // A failed read leaves the publish controls hidden, so say so: without this the dialog
+          // looked complete while quietly offering one control fewer, and the only way to find out
+          // was the network tab. `ngOnInit` is re-entered as a refresh after an access change, so a
+          // value from the previous read may still be here -- drop it rather than leave a state
+          // nothing has confirmed on screen while the toast says it is gone.
           this.isPublic = null;
+          this.publishStatus = undefined;
           this.notificationService.error(
             `Could not read whether this ${this.type} is public, so that choice is not shown.`
           );
         },
       });
+    if (this.type === "workflow") {
+      // The panel describes the saved copy and the editor saves on a debounce, so re-read when a
+      // save lands -- until then the server would still answer with the copy before it.
+      this.workflowPersistService
+        .getWorkflowPersistedStream()
+        .pipe(untilDestroyed(this))
+        .subscribe(() => this.refreshPublishStatus());
+    }
   }
 
   ngOnDestroy(): void {
@@ -350,7 +449,11 @@ export class ShareAccessComponent implements OnInit, OnDestroy {
       const cloneWarning = this.descriptor?.affordances?.clonable ? ", along with the right to clone your work" : "";
       const modal: NzModalRef = this.modalService.create({
         nzTitle: "Notice",
-        nzContent: `Publishing your ${this.type} would grant all Texera users read access to your ${this.type}${cloneWarning}.`,
+        nzContent:
+          `Publishing your ${this.type} would grant all Texera users read access to your ${this.type}${cloneWarning}.` +
+          (this.type === "workflow"
+            ? " The public will follow your latest version as you save it, unless you pin one."
+            : ""),
         nzFooter: [
           {
             label: "Cancel",
@@ -414,6 +517,7 @@ export class ShareAccessComponent implements OnInit, OnDestroy {
           if (this.inWorkspace) {
             this.workflowActionService.setWorkflowIsPublished(published ? 1 : 0);
           }
+          this.refreshPublishStatus();
           this.notificationService.success(`${label} ${published ? "published" : "unpublished"} successfully`);
         },
         error: (error: unknown) => {