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
new file mode 100644
index 00000000000..38174dd1f66
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.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.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 javax.ws.rs.NotFoundException
+import scala.util.Try
+
+/**
+  * Version pinning for public workflows.
+  *
+  * A public workflow follows the author's latest, as publishing has always done, until the author
+  * pins the version they have now: the public then keeps seeing that frozen copy while the author's
+  * later edits stay in the workflow's own columns until they pin again.
+  *
+  * `is_public` stays the on/off switch; `published_content` is the pin, NULL while following. A pin
+  * freezes everything on public show -- the graph, the title, the description and the view it opens
+  * in -- because a copy that froze only its graph would still advertise a title nobody published.
+  *
+  * Not to be confused with sharing: a user granted access always tracks the author's latest, pin or
+  * no pin. Only viewers who arrive because the workflow is public are held at the frozen copy.
+  */
+object WorkflowPublishService extends LazyLogging {
+
+  private def context: DSLContext = SqlServer.getInstance().createDSLContext()
+
+  /**
+    * 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.
+    */
+  case class PublishStatus(
+      isPublished: Boolean,
+      isPinned: Boolean,
+      hasUnpublishedChanges: Boolean
+  )
+
+  /**
+    * Whether two workflow contents describe the same graph. Compared as parsed trees, because the
+    * two blobs travel by different routes and the same graph can come back with its whitespace or
+    * key order rearranged -- reporting that as an edit the public cannot see would be an alarm the
+    * author cannot clear.
+    */
+  private def sameContent(a: String, b: String): Boolean =
+    a == b || Try(objectMapper.readTree(a) == objectMapper.readTree(b)).getOrElse(false)
+
+  /** The workflow, or a 404. */
+  private def requireWorkflow(wid: Integer): Workflow =
+    Option(new WorkflowDao(context.configuration).fetchOneByWid(wid))
+      .getOrElse(throw new NotFoundException(s"Workflow $wid not found"))
+
+  /**
+    * Turns publishing on, and touches nothing else. A workflow coming back from private is
+    * following the author's latest, because unpublishing always drops the pin: coming back should
+    * not silently put old public content back on show. Called on a workflow that is already public
+    * it changes nothing, pin included.
+    */
+  def publish(wid: Integer): PublishStatus = {
+    val updated = context
+      .update(WORKFLOW)
+      .set(WORKFLOW.IS_PUBLIC, java.lang.Boolean.TRUE)
+      .where(WORKFLOW.WID.eq(wid))
+      .execute()
+    if (updated == 0) {
+      throw new NotFoundException(s"Workflow $wid not found")
+    }
+    logger.info(s"Workflow $wid published, following latest")
+    statusOf(wid)
+  }
+
+  /**
+    * Freezes the author's current copy as the public one, and turns publishing on. The title, the
+    * description and the default view freeze with the graph: they are as public as it is, and the
+    * database refuses a pinned copy that carries only part of itself. The view matters because a
+    * form's definition rides inside the content -- serving the live preference over a frozen graph
+    * would open a form on a copy that has none.
+    *
+    * Each column is copied from its own row rather than from a workflow read a moment earlier, so
+    * there is no window in which the author's next save lands and the pin freezes the version
+    * before it -- which would leave them looking at "you have unpublished changes" the instant
+    * after they pinned.
+    *
+    * @return how many rows it matched, so a missing workflow is distinguishable from a done one.
+    */
+  private def writePin(wid: Integer): Int =
+    context
+      .update(WORKFLOW)
+      .set(WORKFLOW.IS_PUBLIC, java.lang.Boolean.TRUE)
+      .set(WORKFLOW.PUBLISHED_CONTENT, WORKFLOW.CONTENT)
+      .set(WORKFLOW.PUBLISHED_NAME, WORKFLOW.NAME)
+      .set(WORKFLOW.PUBLISHED_DESCRIPTION, WORKFLOW.DESCRIPTION)
+      .set(WORKFLOW.PUBLISHED_DEFAULT_VIEW, WORKFLOW.DEFAULT_VIEW)
+      .where(WORKFLOW.WID.eq(wid))
+      .execute()
+
+  /**
+    * Clears the pinned copy in one statement, optionally unpublishing too: the constraint accepts a
+    * 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.
+    *
+    * @return how many rows it matched, so a missing workflow is distinguishable from a done one.
+    */
+  private def clearPin(wid: Integer, alsoUnpublish: Boolean = false): Int = {
+    val cleared = context
+      .update(WORKFLOW)
+      .set(WORKFLOW.PUBLISHED_CONTENT, null.asInstanceOf[String])
+      .set(WORKFLOW.PUBLISHED_NAME, null.asInstanceOf[String])
+      .set(WORKFLOW.PUBLISHED_DESCRIPTION, null.asInstanceOf[String])
+      .set(WORKFLOW.PUBLISHED_DEFAULT_VIEW, null.asInstanceOf[DefaultViewEnum])
+    val statement =
+      if (alsoUnpublish) cleared.set(WORKFLOW.IS_PUBLIC, java.lang.Boolean.FALSE) else cleared
+    statement.where(WORKFLOW.WID.eq(wid)).execute()
+  }
+
+  /** Pins the current content as the public copy. Moving a pin forward is the same operation. */
+  def pinLatest(wid: Integer): PublishStatus = {
+    if (writePin(wid) == 0) {
+      throw new NotFoundException(s"Workflow $wid not found")
+    }
+    logger.info(s"Workflow $wid pinned to its latest content")
+    statusOf(wid)
+  }
+
+  /**
+    * Drops the pin, so the public follows the author's latest again. The workflow stays public.
+    */
+  def unpin(wid: Integer): PublishStatus = {
+    if (clearPin(wid) == 0) {
+      throw new NotFoundException(s"Workflow $wid not found")
+    }
+    logger.info(s"Workflow $wid unpinned, following latest")
+    statusOf(wid)
+  }
+
+  /**
+    * Turns publishing off and drops the pin. Publishing again starts in the following state; the
+    * previous frozen copy is deliberately not remembered, so an unpublish/re-publish cycle cannot
+    * silently restore old public content.
+    */
+  def unpublish(wid: Integer): Unit = {
+    if (clearPin(wid, alsoUnpublish = true) == 0) {
+      throw new NotFoundException(s"Workflow $wid not found")
+    }
+    logger.info(s"Workflow $wid unpublished")
+  }
+
+  /** Whether a version is pinned, and whether it is holding edits back. */
+  def statusOf(wid: Integer): PublishStatus = {
+    val workflow = requireWorkflow(wid)
+    val pinned = workflow.getPublishedContent != null
+    PublishStatus(
+      isPublished = workflow.getIsPublic,
+      isPinned = pinned,
+      // Literally "what the public sees is not what you have". Every field the pin freezes counts:
+      // a rename the public cannot see is held back exactly as an edit to the graph is. Compared on
+      // values rather than on a version id, so that an edit and its undo report nothing held back.
+      hasUnpublishedChanges = pinned && (
+        !sameContent(workflow.getPublishedContent, workflow.getContent) ||
+          workflow.getPublishedName != workflow.getName ||
+          workflow.getPublishedDescription != workflow.getDescription ||
+          workflow.getPublishedDefaultView != workflow.getDefaultView
+      )
+    )
+  }
+}
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 b8bead4b0ea..711839b715c 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
@@ -35,13 +35,14 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
   WorkflowUserAccessDao
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos._
+import org.apache.texera.dao.jooq.generated.tables.records.WorkflowRecord
 import org.apache.texera.service.util.LargeBinaryManager
 import org.apache.texera.web.resource.dashboard.hub.EntityType
 import org.apache.texera.web.service.WarehouseReadGuard
 import org.apache.texera.web.resource.dashboard.hub.HubResource.recordCloneAction
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource._
 import org.jooq.impl.DSL.{noCondition, max}
-import org.jooq.{Condition, DSLContext, Record10, Result, SelectOnConditionStep}
+import org.jooq.{Condition, DSLContext, Record10, Result, SelectOnConditionStep, TableField}
 
 import java.sql.Timestamp
 import java.util
@@ -164,12 +165,21 @@ object WorkflowResource {
     workflow
   }
 
-  private def updateWorkflowField(
-      workflow: Workflow,
+  /**
+    * Writes one field of a workflow, and only that field. The endpoints take a whole `Workflow`
+    * from the client, of which exactly two things are used: which workflow, and the new value.
+    *
+    * Reading the row and writing the whole POJO back would carry every other column with it, so
+    * anything landing between the read and the write was silently rewritten to whatever the read had
+    * seen: a save reverted, a publish undone, or -- since a pin travels in those columns too -- a
+    * workflow the author had just unpublished put back on public show under its frozen copy.
+    */
+  private def updateWorkflowField[T](
+      wid: Integer,
       sessionUser: SessionUser,
-      updateFunction: Workflow => Unit
+      field: TableField[WorkflowRecord, T],
+      value: T
   ): Unit = {
-    val wid = workflow.getWid
     val user = sessionUser.getUser
 
     if (
@@ -178,9 +188,7 @@ object WorkflowResource {
         user.getUid
       )
     ) {
-      val userWorkflow = workflowDao.fetchOneByWid(wid)
-      updateFunction(userWorkflow)
-      workflowDao.update(userWorkflow)
+      context.update(WORKFLOW).set(field, value).where(WORKFLOW.WID.eq(wid)).execute()
     } else {
       throw new ForbiddenException("No sufficient access privilege.")
     }
@@ -490,9 +498,13 @@ class WorkflowResource extends LazyLogging {
 
   /**
     * Persists a plain save by updating only the fields the client sends
-    * (name/description/content/is_public). It deliberately leaves `default_view` untouched --
-    * that column is owned by /set-default-view alone -- so a save can never clobber a
-    * concurrent change. Timestamps are likewise not rewritten here.
+    * (name/description/content). It deliberately leaves `default_view` untouched -- that column is
+    * owned by /set-default-view alone -- so a save can never clobber a concurrent change.
+    * Timestamps are likewise not rewritten here.
+    *
+    * `is_public` is left out for the same reason: publishing is owned by /public and /private, and a
+    * save carrying a stale value would now flip a pinned workflow to private, which the pin
+    * constraint refuses outright -- leaving a workflow that cannot be saved at all.
     */
   private def saveWorkflowFields(workflow: Workflow): Unit = {
     context
@@ -500,7 +512,6 @@ class WorkflowResource extends LazyLogging {
       .set(WORKFLOW.NAME, workflow.getName)
       .set(WORKFLOW.DESCRIPTION, workflow.getDescription)
       .set(WORKFLOW.CONTENT, workflow.getContent)
-      .set(WORKFLOW.IS_PUBLIC, workflow.getIsPublic)
       .where(WORKFLOW.WID.eq(workflow.getWid))
       .execute()
   }
@@ -704,7 +715,7 @@ class WorkflowResource extends LazyLogging {
       workflow: Workflow,
       @Auth sessionUser: SessionUser
   ): Unit = {
-    updateWorkflowField(workflow, sessionUser, _.setName(workflow.getName))
+    updateWorkflowField(workflow.getWid, sessionUser, WORKFLOW.NAME, workflow.getName)
   }
 
   @POST
@@ -716,7 +727,12 @@ class WorkflowResource extends LazyLogging {
       workflow: Workflow,
       @Auth sessionUser: SessionUser
   ): Unit = {
-    updateWorkflowField(workflow, sessionUser, _.setDescription(workflow.getDescription))
+    updateWorkflowField(
+      workflow.getWid,
+      sessionUser,
+      WORKFLOW.DESCRIPTION,
+      workflow.getDescription
+    )
   }
 
   @PUT
@@ -726,9 +742,7 @@ class WorkflowResource extends LazyLogging {
     if (!WorkflowAccessResource.hasWriteAccess(wid, user.getUid)) {
       throw new ForbiddenException(s"You do not have permission to modify workflow $wid")
     }
-    val workflow: Workflow = workflowDao.fetchOneByWid(wid)
-    workflow.setIsPublic(true)
-    workflowDao.update(workflow)
+    WorkflowPublishService.publish(wid)
   }
 
   @PUT
@@ -738,9 +752,66 @@ class WorkflowResource extends LazyLogging {
     if (!WorkflowAccessResource.hasWriteAccess(wid, user.getUid)) {
       throw new ForbiddenException(s"You do not have permission to modify workflow $wid")
     }
-    val workflow: Workflow = workflowDao.fetchOneByWid(wid)
-    workflow.setIsPublic(false)
-    workflowDao.update(workflow)
+    WorkflowPublishService.unpublish(wid)
+  }
+
+  /**
+    * Pins the author's current version as the public copy, so later edits stop reaching the public.
+    * Also how a pin moves forward, which is the only way edits become public while one is in place.
+    */
+  @POST
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @RolesAllowed(Array("REGULAR", "ADMIN"))
+  @Path("/pin/{wid}")
+  def pinLatest(
+      @PathParam("wid") wid: Integer,
+      @Auth user: SessionUser
+  ): WorkflowPublishService.PublishStatus = {
+    requirePublishable(wid, user)
+    WorkflowPublishService.pinLatest(wid)
+  }
+
+  /**
+    * Drops the pin, so the public follows the author's latest again. Guarded on write access, the
+    * same as pinning: whoever may pin may undo it.
+    */
+  @DELETE
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @RolesAllowed(Array("REGULAR", "ADMIN"))
+  @Path("/pin/{wid}")
+  def unpin(
+      @PathParam("wid") wid: Integer,
+      @Auth user: SessionUser
+  ): WorkflowPublishService.PublishStatus = {
+    requirePublishable(wid, user)
+    WorkflowPublishService.unpin(wid)
+  }
+
+  /** What the share dialog's publish panel reads: published, pinned, and holding edits back. */
+  @GET
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @RolesAllowed(Array("REGULAR", "ADMIN"))
+  @Path("/publish-status/{wid}")
+  def getPublishStatus(
+      @PathParam("wid") wid: Integer,
+      @Auth user: SessionUser
+  ): WorkflowPublishService.PublishStatus = {
+    // Write access rather than read: whether edits are being held back is nobody else's business.
+    requireWriteAccess(wid, user)
+    WorkflowPublishService.statusOf(wid)
+  }
+
+  private def requireWriteAccess(wid: Integer, user: SessionUser): Unit =
+    if (!WorkflowAccessResource.hasWriteAccess(wid, user.getUid)) {
+      throw new ForbiddenException(s"You do not have permission to modify workflow $wid")
+    }
+
+  /** What the pin endpoints need: writable by this user, and published in the first place. */
+  private def requirePublishable(wid: Integer, user: SessionUser): Unit = {
+    requireWriteAccess(wid, user)
+    if (!WorkflowAccessResource.isPublic(wid)) {
+      throw new BadRequestException(s"Workflow $wid is not published")
+    }
   }
 
   /**
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
new file mode 100644
index 00000000000..588c4704d2c
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
@@ -0,0 +1,564 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.resource.dashboard.user.workflow
+
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW, WORKFLOW_USER_ACCESS}
+import org.apache.texera.dao.jooq.generated.enums.{DefaultViewEnum, PrivilegeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.daos.{
+  UserDao,
+  WorkflowDao,
+  WorkflowUserAccessDao
+}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{User, Workflow, WorkflowUserAccess}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import org.jooq.{ExecuteContext, ExecuteListener}
+import org.jooq.impl.{DefaultConfiguration, DefaultExecuteListenerProvider}
+
+import java.time.OffsetDateTime
+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.
+  *
+  * Only the state itself is covered here: nothing serves the pinned copy to a reader yet, so the
+  * assertions are about which copy each operation leaves stored.
+  */
+class WorkflowPublishSpec
+    extends AnyFlatSpec
+    with BeforeAndAfterAll
+    with Matchers
+    with MockTexeraDB {
+
+  private val exampleCreationTime = OffsetDateTime.parse("2025-01-01T00:00:00Z")
+
+  private def makeUser(uid: Int, name: String): User = {
+    val user = new User
+    user.setUid(Integer.valueOf(uid))
+    user.setName(name)
+    user.setEmail(s"$name@example.com")
+    user.setRole(UserRoleEnum.ADMIN)
+    user.setComment("test")
+    user.setAccountCreationTime(exampleCreationTime)
+    user
+  }
+
+  /** The author. */
+  private val owner = makeUser(1, "publish_owner")
+
+  /** A stranger: no access of their own, so nothing about this workflow is theirs to change. */
+  private val stranger = makeUser(2, "publish_stranger")
+
+  private val ownerSession = new SessionUser(owner)
+  private val strangerSession = new SessionUser(stranger)
+
+  private val workflowResource = new WorkflowResource()
+
+  private val publishedContent = """{"operators":[],"note":"content_as_published"}"""
+  private val editedContent = """{"operators":[],"note":"content_only_a_draft"}"""
+
+  private def workflowDao = new WorkflowDao(getDSLContext.configuration())
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    val userDao = new UserDao(getDSLContext.configuration())
+    userDao.insert(owner)
+    userDao.insert(stranger)
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  /** Creates a workflow owned by `owner` holding [[publishedContent]]. */
+  private def createWorkflow(name: String): Integer = {
+    val workflow = new Workflow()
+    workflow.setName(name)
+    workflow.setDescription("a workflow")
+    workflow.setContent(publishedContent)
+    workflowResource.createWorkflow(workflow, ownerSession).workflow.getWid
+  }
+
+  /**
+    * Publishes and pins in one step, which is the state most of these tests are about. Publishing on
+    * its own leaves the workflow following the author's latest; pinning is what freezes a copy.
+    */
+  private def publishPinned(wid: Integer): WorkflowPublishService.PublishStatus = {
+    workflowResource.makePublic(wid, ownerSession)
+    workflowResource.pinLatest(wid, ownerSession)
+  }
+
+  /** Saves `content` as the author's working copy, the way an autosave would. */
+  private def edit(wid: Integer, content: String): Unit = {
+    val workflow = workflowDao.fetchOneByWid(wid)
+    workflow.setContent(content)
+    workflowResource.persistWorkflow(workflow, ownerSession)
+  }
+
+  /** Renames and re-describes the author's working copy, the way the dashboard does. */
+  private def relabel(wid: Integer, name: String, description: String): Unit = {
+    val workflow = workflowDao.fetchOneByWid(wid)
+    workflow.setName(name)
+    workflow.setDescription(description)
+    workflowResource.persistWorkflow(workflow, ownerSession)
+  }
+
+  /**
+    * Runs `interleaved` in the last moment before `act` sends its own write, which is where a second
+    * request slips in unnoticed. Driven off the statement itself rather than off a thread, so the
+    * ordering is the same on every run.
+    */
+  private def interleaving(interleaved: () => Unit)(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 = {
+        // The workflow table itself, not workflow_version or the access tables: matching those too
+        // would let a later change to one of these paths interleave at the wrong moment and leave
+        // the test passing for the wrong reason.
+        val sql = Option(ctx.sql()).getOrElse("").toLowerCase
+        if (pending && sql.startsWith("update") && sql.contains("\"workflow\" set")) {
+          pending = false
+          interleaved()
+        }
+      }
+    }))
+    try act
+    finally configuration.set(previousListeners: _*)
+    withClue("nothing was interleaved, so this proves nothing: ") { pending shouldBe false }
+  }
+
+  private def statusOf(wid: Integer): WorkflowPublishService.PublishStatus =
+    workflowResource.getPublishStatus(wid, ownerSession)
+
+  /** Grants `stranger` explicit access, which makes them a collaborator rather than an outsider. */
+  private def grantAccess(wid: Integer, privilege: PrivilegeEnum): Unit =
+    new WorkflowUserAccessDao(getDSLContext.configuration())
+      .insert(new WorkflowUserAccess(stranger.getUid, wid, privilege))
+
+  private def revokeAccess(wid: Integer): Unit =
+    getDSLContext
+      .deleteFrom(WORKFLOW_USER_ACCESS)
+      .where(WORKFLOW_USER_ACCESS.WID.eq(wid).and(WORKFLOW_USER_ACCESS.UID.eq(stranger.getUid)))
+      .execute()
+
+  behavior of "publishing"
+
+  it should "follow the author's latest by default" in {
+    val wid = createWorkflow("publish_follows_latest")
+    workflowResource.makePublic(wid, ownerSession)
+
+    val status = statusOf(wid)
+    status.isPublished shouldBe true
+    status.isPinned shouldBe false
+    // Nothing is frozen, so nothing is held back however much the author edits.
+    status.hasUnpublishedChanges shouldBe false
+    workflowDao.fetchOneByWid(wid).getPublishedContent shouldBe null
+
+    edit(wid, editedContent)
+    statusOf(wid).hasUnpublishedChanges shouldBe false
+  }
+
+  it should "pin the current version as the public copy" in {
+    val wid = createWorkflow("pins_current_version")
+    val status = publishPinned(wid)
+
+    status.isPublished shouldBe true
+    status.isPinned shouldBe true
+    status.hasUnpublishedChanges shouldBe false
+    workflowDao.fetchOneByWid(wid).getPublishedContent shouldBe publishedContent
+  }
+
+  it should "pin a workflow that has no description" in {
+    // description is nullable and the constraint does not ask for published_description, so a
+    // workflow saved without one has to pin like any other rather than fail on the way in.
+    val workflow = new Workflow()
+    workflow.setName("pins_without_a_description")
+    workflow.setContent(publishedContent)
+    val wid = workflowResource.createWorkflow(workflow, ownerSession).workflow.getWid
+
+    val status = publishPinned(wid)
+
+    status.isPinned shouldBe true
+    status.hasUnpublishedChanges shouldBe false
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getDescription shouldBe null
+    stored.getPublishedDescription shouldBe null
+
+    // ...and writing one afterwards is an unpublished change like any other.
+    relabel(wid, "pins_without_a_description", "described later")
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "follow the author's latest again once the pin is dropped" in {
+    val wid = createWorkflow("unpin_follows_latest")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    val status = workflowResource.unpin(wid, ownerSession)
+
+    status.isPublished shouldBe true
+    status.isPinned shouldBe false
+    status.hasUnpublishedChanges shouldBe false
+    // Still public; only the frozen copy is gone.
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe null
+  }
+
+  it should "freeze the default view with the copy" in {
+    // The form's definition rides inside the content, so pinning a canvas version and then switching
+    // the workflow to the form view would otherwise leave the public opening a form that the frozen
+    // copy does not contain.
+    val wid = createWorkflow("view_freezes_with_the_copy")
+    publishPinned(wid)
+
+    getDSLContext
+      .update(WORKFLOW)
+      .set(WORKFLOW.DEFAULT_VIEW, DefaultViewEnum.FORM)
+      .where(WORKFLOW.WID.eq(wid))
+      .execute()
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getDefaultView shouldBe DefaultViewEnum.FORM
+    stored.getPublishedDefaultView shouldBe DefaultViewEnum.CANVAS
+  }
+
+  it should "clear the frozen default view when the pin is dropped" in {
+    val wid = createWorkflow("view_clears_with_the_pin")
+    publishPinned(wid)
+    workflowDao.fetchOneByWid(wid).getPublishedDefaultView shouldBe DefaultViewEnum.CANVAS
+
+    workflowResource.unpin(wid, ownerSession)
+
+    workflowDao.fetchOneByWid(wid).getPublishedDefaultView shouldBe null
+  }
+
+  it should "leave the pinned copy untouched when the author edits afterwards" in {
+    val wid = createWorkflow("edit_stays_private")
+    publishPinned(wid)
+
+    edit(wid, editedContent)
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    // The author's own working copy has moved on...
+    stored.getContent shouldBe editedContent
+    // ...but the copy that was frozen has not.
+    stored.getPublishedContent shouldBe publishedContent
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "pin the save that lands while it is pinning, not the version before it" in {
+    // 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.
+    val wid = createWorkflow("pin_takes_the_row_as_it_stands")
+    workflowResource.makePublic(wid, ownerSession)
+
+    interleaving(() => edit(wid, editedContent)) {
+      workflowResource.pinLatest(wid, ownerSession)
+    }
+
+    workflowDao.fetchOneByWid(wid).getPublishedContent shouldBe editedContent
+    statusOf(wid).hasUnpublishedChanges shouldBe false
+  }
+
+  it should "move the pin forward to the author's current version" in {
+    val wid = createWorkflow("repin_updates_public")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    val status = workflowResource.pinLatest(wid, ownerSession)
+
+    status.isPinned shouldBe true
+    status.hasUnpublishedChanges shouldBe false
+    workflowDao.fetchOneByWid(wid).getPublishedContent shouldBe editedContent
+  }
+
+  it should "count a rename as an unpublished change" in {
+    // The pin freezes the title too, so the public is still being shown the old one -- the panel has
+    // to say so, or the author reads "nothing held back" while the hub disagrees with their editor.
+    val wid = createWorkflow("rename_counts_as_drift")
+    publishPinned(wid)
+    statusOf(wid).hasUnpublishedChanges shouldBe false
+
+    relabel(wid, "renamed_after_pinning", "a workflow")
+
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "count a description edit as an unpublished change" in {
+    val wid = createWorkflow("description_counts_as_drift")
+    publishPinned(wid)
+
+    relabel(wid, "description_counts_as_drift", "rewritten after pinning")
+
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "count a change of view as an unpublished change" in {
+    val wid = createWorkflow("view_counts_as_drift")
+    publishPinned(wid)
+
+    getDSLContext
+      .update(WORKFLOW)
+      .set(WORKFLOW.DEFAULT_VIEW, DefaultViewEnum.FORM)
+      .where(WORKFLOW.WID.eq(wid))
+      .execute()
+
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "report no unpublished changes when an edit is undone" in {
+    val wid = createWorkflow("undo_clears_badge")
+    publishPinned(wid)
+
+    edit(wid, editedContent)
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+
+    edit(wid, publishedContent)
+    statusOf(wid).hasUnpublishedChanges shouldBe false
+  }
+
+  it should "report no unpublished changes when the same graph comes back rearranged" in {
+    // The two copies travel by different routes, and the editor is free to hand back the same graph
+    // with its keys in another order. Reporting that as an edit is an alarm the author cannot clear.
+    val wid = createWorkflow("reformat_is_not_an_edit")
+    publishPinned(wid)
+
+    edit(wid, """{ "note":"content_as_published",  "operators": [] }""")
+
+    statusOf(wid).hasUnpublishedChanges shouldBe false
+  }
+
+  it should "do nothing when unpinning a workflow that is following" in {
+    // The endpoint is reachable whatever the dialog shows, and asking for the state it is already in
+    // is not an error -- it just has nothing to clear.
+    val wid = createWorkflow("unpin_while_following")
+    workflowResource.makePublic(wid, ownerSession)
+
+    val status = workflowResource.unpin(wid, ownerSession)
+
+    status.isPublished shouldBe true
+    status.isPinned shouldBe false
+    workflowDao.fetchOneByWid(wid).getIsPublic shouldBe true
+  }
+
+  it should "leave a pin alone when the workflow is published again" in {
+    // Publishing is an on/off switch and this one is already on, so it has nothing to turn: the
+    // frozen copy is not quietly dropped underneath a public that is reading it.
+    val wid = createWorkflow("republish_keeps_the_pin")
+    publishPinned(wid)
+
+    workflowResource.makePublic(wid, ownerSession)
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe publishedContent
+    statusOf(wid).isPinned shouldBe true
+  }
+
+  it should "drop the pinned copy on unpublish" in {
+    val wid = createWorkflow("unpublish_clears_pin")
+    publishPinned(wid)
+
+    workflowResource.makePrivate(wid, ownerSession)
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getIsPublic shouldBe false
+    stored.getPublishedContent shouldBe null
+  }
+
+  it should "not resurrect the previous pin after unpublish and re-publish" in {
+    val wid = createWorkflow("unpublish_then_publish")
+    publishPinned(wid)
+    edit(wid, editedContent)
+    workflowResource.makePrivate(wid, ownerSession)
+
+    // Publishing again starts in the following state; the copy that used to be public is gone.
+    workflowResource.makePublic(wid, ownerSession)
+
+    statusOf(wid).isPinned shouldBe false
+    workflowDao.fetchOneByWid(wid).getPublishedContent shouldBe null
+  }
+
+  it should "publish a workflow that is created already public" in {
+    val workflow = new Workflow()
+    workflow.setName("created_public")
+    workflow.setDescription("a workflow")
+    workflow.setContent(publishedContent)
+    workflow.setIsPublic(true)
+    val wid = workflowResource.createWorkflow(workflow, ownerSession).workflow.getWid
+
+    // Asking for a public workflow up front lands in the same following state as any other new
+    // public workflow, rather than being pinned by surprise.
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe null
+  }
+
+  it should "reject publishing by a user without write access" in {
+    val wid = createWorkflow("publish_requires_write")
+    a[ForbiddenException] should be thrownBy workflowResource.makePublic(wid, strangerSession)
+  }
+
+  it should "refuse to pin, unpin or report status without write access" in {
+    val wid = createWorkflow("pin_requires_write")
+    publishPinned(wid)
+    a[ForbiddenException] should be thrownBy workflowResource.pinLatest(wid, strangerSession)
+    a[ForbiddenException] should be thrownBy workflowResource.unpin(wid, strangerSession)
+    a[ForbiddenException] should be thrownBy workflowResource.getPublishStatus(wid, strangerSession)
+  }
+
+  it should "reject pinning and unpinning a workflow that is not published" in {
+    val wid = createWorkflow("pin_requires_published")
+    a[BadRequestException] should be thrownBy workflowResource.pinLatest(wid, ownerSession)
+    a[BadRequestException] should be thrownBy workflowResource.unpin(wid, ownerSession)
+  }
+
+  it should "answer 404 for every operation on a workflow that does not exist" in {
+    // Asked of the service rather than the endpoints: a missing workflow has no access row either,
+    // so the endpoints answer 403 first and never reach these. 404 is the service's own contract.
+    val missing = Integer.valueOf(987654)
+    a[NotFoundException] should be thrownBy WorkflowPublishService.publish(missing)
+    a[NotFoundException] should be thrownBy WorkflowPublishService.pinLatest(missing)
+    a[NotFoundException] should be thrownBy WorkflowPublishService.unpin(missing)
+    a[NotFoundException] should be thrownBy WorkflowPublishService.unpublish(missing)
+    a[NotFoundException] should be thrownBy WorkflowPublishService.statusOf(missing)
+  }
+
+  behavior of "saving a published workflow"
+
+  it should "not roll back a publish that lands while a save is in flight" in {
+    // A save used to carry `is_public` along. An editor open since before the workflow was
+    // published holds a snapshot saying private, and saving it put that back -- taking a pinned
+    // workflow private underneath its own frozen copy, which the database refuses outright, so the
+    // author was left with an editor that could no longer save. The save no longer names the column.
+    val wid = createWorkflow("save_cannot_roll_back_publish")
+
+    // The snapshot an editor opened before any of this was published.
+    val stale = workflowDao.fetchOneByWid(wid)
+    stale.getIsPublic shouldBe false
+
+    publishPinned(wid)
+
+    stale.setContent("""{"operators":[],"note":"from_a_stale_client"}""")
+    workflowResource.persistWorkflow(stale, ownerSession)
+
+    // The save went through, and it moved the working copy only.
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getContent shouldBe """{"operators":[],"note":"from_a_stale_client"}"""
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe publishedContent
+  }
+
+  it should "not let a save change the publish state" in {
+    val wid = createWorkflow("save_cannot_publish")
+    publishPinned(wid)
+
+    // A stale or hostile client sending the whole POJO back with the publish columns rewritten.
+    val tampered = workflowDao.fetchOneByWid(wid)
+    tampered.setContent(editedContent)
+    tampered.setIsPublic(false)
+    tampered.setPublishedContent(editedContent)
+    workflowResource.persistWorkflow(tampered, ownerSession)
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe publishedContent
+  }
+
+  it should "not let a collaborator's save change the publish state" in {
+    val wid = createWorkflow("collaborator_cannot_publish")
+    publishPinned(wid)
+    grantAccess(wid, PrivilegeEnum.WRITE)
+
+    try {
+      val tampered = workflowDao.fetchOneByWid(wid)
+      tampered.setContent(editedContent)
+      tampered.setIsPublic(false)
+      tampered.setPublishedContent(editedContent)
+      workflowResource.persistWorkflow(tampered, strangerSession)
+
+      val stored = workflowDao.fetchOneByWid(wid)
+      stored.getIsPublic shouldBe true
+      stored.getPublishedContent shouldBe publishedContent
+    } finally revokeAccess(wid)
+  }
+
+  it should "not let a rename undo a publish that lands first" in {
+    // A rename used to read the whole row and write it all back, so a publish landing in that window
+    // was reverted to what the read had seen: the author pressed Public, was told it worked, and the
+    // workflow was private again.
+    val wid = createWorkflow("rename_cannot_undo_publish")
+
+    interleaving(() => publishPinned(wid)) {
+      val body = new Workflow()
+      body.setWid(wid)
+      body.setName("renamed_during_a_publish")
+      workflowResource.updateWorkflowName(body, ownerSession)
+    }
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getName shouldBe "renamed_during_a_publish"
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe publishedContent
+  }
+
+  it should "not let a rename put an unpublished workflow back on show" in {
+    // The same window, the other way round, and the one that matters: the author takes the workflow
+    // down, and a rename in flight restores the row as it was -- public, still carrying the frozen
+    // copy the public had been reading.
+    val wid = createWorkflow("rename_cannot_republish")
+    publishPinned(wid)
+
+    interleaving(() => workflowResource.makePrivate(wid, ownerSession)) {
+      val body = new Workflow()
+      body.setWid(wid)
+      body.setName("renamed_during_an_unpublish")
+      workflowResource.updateWorkflowName(body, ownerSession)
+    }
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getName shouldBe "renamed_during_an_unpublish"
+    stored.getIsPublic shouldBe false
+    stored.getPublishedContent shouldBe null
+  }
+
+  it should "not let a rename change the publish state" in {
+    val wid = createWorkflow("rename_cannot_publish")
+    publishPinned(wid)
+
+    val tampered = workflowDao.fetchOneByWid(wid)
+    tampered.setName("renamed")
+    tampered.setIsPublic(false)
+    tampered.setPublishedContent(editedContent)
+    workflowResource.updateWorkflowName(tampered, ownerSession)
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getName shouldBe "renamed"
+    stored.getIsPublic shouldBe true
+    stored.getPublishedContent shouldBe publishedContent
+  }
+}