From f8b02912489645ec493a1811ca0350a34ed69bfd Mon Sep 17 00:00:00 2001
From: Yang Zhang <yangz75@uci.edu>
Date: Fri, 21 Aug 2026 23:11:52 -0700
Subject: [PATCH 1/3] feat(workflow): add the published-copy columns

A public workflow follows the author's latest content today: every save reaches
the Hub immediately. Pinning a version as the public copy needs somewhere to keep
that copy, which is what these columns are.

`is_public` stays the on/off switch. `published_content` is the pin: NULL means
the workflow follows the author's latest, which is what every workflow does today,
so the migration changes nothing anyone can see. `published_name` and
`published_description` travel with it because a pin has to hold everything on
show, and `workflow_version` stores no metadata at all -- only content deltas.
`published_version_id` names the version row holding that copy, so the revision
panel can mark it and the author can restore it.

The copy is materialized rather than replayed from `workflow_version` because
those rows are reverse JSON-Patch deltas: serving a pinned workflow would mean
folding every newer patch back from the author's current content on each public
read, and a computed value is something the fulltext index cannot cover.

A CHECK constraint makes "private but pinned" unrepresentable, and a PGroonga
index mirrors the latest-content one so public search can match the frozen copy.

Adding columns changes the arity of the generated positional constructor, so the
three copy-producing paths (clone, duplicate, restore-a-version) now build their
POJO with setters -- which is also what stops a later column from silently
shifting a null into the wrong field.

Part of #7828.
---
 .../user/workflow/WorkflowResource.scala      | 31 ++++---
 .../workflow/WorkflowVersionResource.scala    | 11 +--
 .../workflow/PublishedCopySchemaSpec.scala    | 93 +++++++++++++++++++
 sql/changelog.xml                             |  5 +
 sql/texera_ddl.sql                            | 31 ++++++-
 sql/updates/41.sql                            | 93 +++++++++++++++++++
 6 files changed, 244 insertions(+), 20 deletions(-)
 create mode 100644 amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/PublishedCopySchemaSpec.scala
 create mode 100644 sql/updates/41.sql

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 2d438e8dc75..e7fecc06c07 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
@@ -143,6 +143,21 @@ object WorkflowResource {
 
   case class WorkflowIDs(wids: List[Integer], pid: Option[Integer])
 
+  /**
+    * A workflow POJO for the copy-producing paths (clone, duplicate, restore-a-version).
+    *
+    * Built with setters rather than the positional constructor, so that adding a column cannot
+    * silently shift a null into the wrong field -- as adding the published-copy columns would.
+    */
+  def newUnpublishedWorkflow(name: String, description: String, content: String): Workflow = {
+    val workflow = new Workflow()
+    workflow.setName(name)
+    workflow.setDescription(description)
+    workflow.setContent(content)
+    workflow.setIsPublic(false)
+    workflow
+  }
+
   private def updateWorkflowField(
       workflow: Workflow,
       sessionUser: SessionUser,
@@ -504,14 +519,10 @@ class WorkflowResource extends LazyLogging {
         for (wid <- workflowIDs.wids) {
           val oldWorkflow: Workflow = workflowDao.fetchOneByWid(wid)
           val newWorkflow = createWorkflow(
-            new Workflow(
-              null,
+            newUnpublishedWorkflow(
               oldWorkflow.getName + "_copy",
               oldWorkflow.getDescription,
-              assignNewOperatorIds(oldWorkflow.getContent),
-              null,
-              null,
-              false
+              assignNewOperatorIds(oldWorkflow.getContent)
             ),
             sessionUser
           )
@@ -554,14 +565,10 @@ class WorkflowResource extends LazyLogging {
     }
     val oldWorkflow: Workflow = workflowDao.fetchOneByWid(wid)
     val newWorkflow: DashboardWorkflow = createWorkflow(
-      new Workflow(
-        null,
+      newUnpublishedWorkflow(
         oldWorkflow.getName + "_clone",
         oldWorkflow.getDescription,
-        assignNewOperatorIds(oldWorkflow.getContent),
-        null,
-        null,
-        false
+        assignNewOperatorIds(oldWorkflow.getContent)
       ),
       sessionUser
     )
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 e0664b7c1d4..de1f880a834 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
@@ -30,7 +30,8 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{WorkflowDao, WorkflowVe
 import org.apache.texera.dao.jooq.generated.tables.pojos.{Workflow, WorkflowVersion}
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource.{
   DashboardWorkflow,
-  assignNewOperatorIds
+  assignNewOperatorIds,
+  newUnpublishedWorkflow
 }
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowVersionResource._
 import org.jooq.DSLContext
@@ -428,14 +429,10 @@ class WorkflowVersionResource {
     val newWorkflow: DashboardWorkflow =
       try {
         workflowResource.createWorkflow(
-          new Workflow(
-            null,
+          newUnpublishedWorkflow(
             newWorkflowName,
             workflowVersion.getDescription,
-            assignNewOperatorIds(workflowVersion.getContent),
-            null,
-            null,
-            false
+            assignNewOperatorIds(workflowVersion.getContent)
           ),
           sessionUser
         )
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/PublishedCopySchemaSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/PublishedCopySchemaSpec.scala
new file mode 100644
index 00000000000..08a87869c56
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/PublishedCopySchemaSpec.scala
@@ -0,0 +1,93 @@
+/*
+ * 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.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowDao
+import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
+import org.jooq.exception.DataAccessException
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+  * The columns a pinned public copy lives in, and the constraint that keeps them honest.
+  *
+  * Nothing writes them yet: this covers what the migration alone guarantees -- that a workflow
+  * public today keeps behaving as it does, and that a private workflow can never carry a pin.
+  */
+class PublishedCopySchemaSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with MockTexeraDB {
+
+  private var workflowDao: WorkflowDao = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    workflowDao = new WorkflowDao(getDSLContext.configuration())
+  }
+
+  /** A workflow as it exists before anything pins it: content only, nothing frozen. */
+  private def insertWorkflow(name: String, isPublic: Boolean): Workflow = {
+    val workflow = new Workflow()
+    workflow.setName(name)
+    workflow.setDescription("a workflow")
+    workflow.setContent("""{"operators":[]}""")
+    workflow.setIsPublic(isPublic)
+    workflowDao.insert(workflow)
+    workflowDao.fetchOneByWid(workflow.getWid)
+  }
+
+  behavior of "the published-copy columns"
+
+  it should "leave every workflow following the author's latest" in {
+    // The migration adds columns and no backfill, so a workflow that was public before it ran shows
+    // exactly what it showed: nothing is frozen, which is the state the rest of the feature calls
+    // "following".
+    val stored = insertWorkflow("migration_changes_nothing", isPublic = true)
+
+    stored.getPublishedContent shouldBe null
+    stored.getPublishedName shouldBe null
+    stored.getPublishedDescription shouldBe null
+    stored.getPublishedVersionId shouldBe null
+  }
+
+  it should "let a public workflow carry a frozen copy" in {
+    val stored = insertWorkflow("public_may_be_pinned", isPublic = true)
+    stored.setPublishedContent("""{"operators":[]}""")
+    stored.setPublishedName("frozen name")
+    stored.setPublishedDescription("frozen description")
+
+    workflowDao.update(stored)
+
+    workflowDao.fetchOneByWid(stored.getWid).getPublishedName shouldBe "frozen name"
+  }
+
+  it should "refuse a private workflow that carries a frozen copy" in {
+    // A pin only means something while the workflow is public. The database rejects the other case,
+    // so no code path can leave one behind -- unpublishing has to clear the copy.
+    val stored = insertWorkflow("private_cannot_be_pinned", isPublic = false)
+    stored.setPublishedContent("""{"operators":[]}""")
+
+    a[DataAccessException] should be thrownBy workflowDao.update(stored)
+  }
+}
diff --git a/sql/changelog.xml b/sql/changelog.xml
index 3debbad196e..2f4a389bf6c 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -114,6 +114,11 @@
         <sqlFile path="sql/updates/40.sql"/>
     </changeSet>
 
+    <!-- Add version-pinning columns to workflow -->
+    <changeSet id="41" author="yangz75">
+        <sqlFile path="sql/updates/41.sql"/>
+    </changeSet>
+
     <!-- example changeSet
     <changeSet id="1" author="author">
         <sqlFile path="sql/updates/1.sql"/>
diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index c016817a39d..392ddb89e15 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -164,9 +164,28 @@ CREATE TABLE IF NOT EXISTS workflow
     content            TEXT NOT NULL,
     creation_time      TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     last_modified_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
-    is_public          BOOLEAN NOT NULL DEFAULT false
+    is_public          BOOLEAN NOT NULL DEFAULT false,
+    -- is_public is the on/off switch; published_content is the pin. NULL means the public follows the
+    -- author's latest content, non-NULL is the frozen copy the public sees instead. Materialized
+    -- rather than reconstructed from workflow_version, whose rows are reverse deltas.
+    -- published_version_id names the version row holding that copy, which is what the revision panel
+    -- marks so the author can restore it.
+    published_version_id  INT,
+    published_content     TEXT,
+    published_name        VARCHAR(128),
+    published_description TEXT
     );
 
+-- A pin only means something while the workflow is public; these columns are only ever written by
+-- WorkflowPublishService, which clears them on unpublish.
+ALTER TABLE workflow
+    DROP CONSTRAINT IF EXISTS workflow_published_consistent;
+ALTER TABLE workflow
+    DROP CONSTRAINT IF EXISTS workflow_pin_requires_public;
+ALTER TABLE workflow
+    ADD CONSTRAINT workflow_pin_requires_public
+        CHECK (published_content IS NULL OR is_public);
+
 -- workflow_of_user
 CREATE TABLE IF NOT EXISTS workflow_of_user
 (
@@ -675,6 +694,16 @@ BEGIN
       r.tablename, r.tablename, r.index_column, stem_filter
     );
   END LOOP;
+
+  -- Public search matches the pinned copy -- name, description and content together -- rather than
+  -- the author's live values, so it needs its own index alongside idx_workflow_pgroonga. The
+  -- expression must match the one the query builds; see WorkflowSearchQueryBuilder.onVisibleCopy.
+  EXECUTE format(
+    'CREATE INDEX idx_workflow_published_pgroonga ON workflow USING pgroonga ' ||
+    '((COALESCE(published_name, '''') || '' '' || COALESCE(published_description, '''') || '' '' || COALESCE(published_content, ''''))) ' ||
+    'WITH (tokenizer = ''TokenMecab''%s);',
+    stem_filter
+  );
 END $$;
 
 -- END Fulltext search index creation (DO NOT EDIT THIS LINE)
diff --git a/sql/updates/41.sql b/sql/updates/41.sql
new file mode 100644
index 00000000000..ea06ba2b153
--- /dev/null
+++ b/sql/updates/41.sql
@@ -0,0 +1,93 @@
+/*
+ * 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.
+ */
+
+\c texera_db
+
+SET search_path TO texera_db;
+
+BEGIN;
+
+-- Version pinning: a public workflow follows the author's latest until they pin the version they
+-- have now, after which the public keeps seeing that frozen copy. is_public stays the on/off switch;
+-- published_content is the pin, NULL while following. Materialized rather than replayed from
+-- workflow_version, whose rows are reverse deltas that no fulltext index can cover.
+ALTER TABLE workflow
+    -- The version row holding the pinned copy. Its delta is the identity patch, so replaying it
+    -- returns exactly what is on public show; the revision panel marks that row, which is how the
+    -- author restores the public version into their editor.
+    ADD COLUMN IF NOT EXISTS published_version_id  INT,
+    ADD COLUMN IF NOT EXISTS published_content     TEXT,
+    ADD COLUMN IF NOT EXISTS published_name        VARCHAR(128),
+    ADD COLUMN IF NOT EXISTS published_description TEXT;
+
+-- No backfill. Every workflow that is public today has no pin, which is the following state, which is
+-- exactly what it does today: deploying this migration changes nothing anyone can see. Pinning is
+-- something an author opts into afterwards.
+
+-- A pin only means something while the workflow is public, so a private workflow must not carry one.
+-- Making that unrepresentable is cheaper than catching it: unpublishing clears the pin, and no other
+-- path writes these columns.
+DO
+$$
+    BEGIN
+        ALTER TABLE workflow DROP CONSTRAINT IF EXISTS workflow_published_consistent;
+        IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'workflow_pin_requires_public') THEN
+            ALTER TABLE workflow
+                ADD CONSTRAINT workflow_pin_requires_public
+                    CHECK (published_content IS NULL OR is_public);
+        END IF;
+    END
+$$;
+
+COMMIT;
+
+-- Fulltext index over the pinned copy, mirroring the latest-content index built in texera_ddl.sql.
+-- Public search matches a pinned workflow against its pinned name, description and content rather
+-- than the live ones, so those three need an index of their own; unpinned rows keep using the
+-- latest-content index. The expression has to match the one the query builds, or the planner cannot
+-- use it -- see WorkflowSearchQueryBuilder.onVisibleCopy.
+-- Runs outside the transaction above because the plugin probe issues its own commands.
+DO
+$$
+    DECLARE
+        stem_filter   TEXT := '';
+        plugin_status TEXT;
+    BEGIN
+        DROP INDEX IF EXISTS idx_workflow_published_pgroonga;
+
+        WITH plugin_registration AS (SELECT pgroonga_command('plugin_register token_filters/stem') AS result)
+        SELECT CASE
+                   WHEN result::jsonb @> '[true]' THEN 'Plugin registered successfully'
+                   ELSE 'Plugin registration failed'
+                   END
+        INTO plugin_status
+        FROM plugin_registration;
+
+        IF plugin_status = 'Plugin registered successfully' THEN
+            stem_filter := ', plugins=''token_filters/stem'', token_filters=''TokenFilterStem''';
+        END IF;
+
+        EXECUTE format(
+                'CREATE INDEX idx_workflow_published_pgroonga ON workflow USING pgroonga ' ||
+                '((COALESCE(published_name, '''') || '' '' || COALESCE(published_description, '''') || '' '' || COALESCE(published_content, ''''))) ' ||
+                'WITH (tokenizer = ''TokenMecab''%s);',
+                stem_filter
+                );
+    END
+$$;

From 8f56d11635f4ba6a43b8bb20a8c25f140e690fec Mon Sep 17 00:00:00 2001
From: Yang Zhang <yangz75@uci.edu>
Date: Sat, 22 Aug 2026 14:35:59 -0700
Subject: [PATCH 2/3] feat(workflow): freeze a pinned version as the public
 copy

A public workflow follows the author's latest content, as publishing has
always done. This adds the other state: the author pins the version they
have now, and the public copy stops moving until they pin again.

`is_public` stays the on/off switch; `published_content` is the pin, NULL
while following. `WorkflowPublishService` owns the two states, and three
endpoints expose them: POST and DELETE `/workflow/pin/{wid}` to pin and
unpin, GET `/workflow/publish-status/{wid}` for what the author is shown.
Publishing and unpublishing move through the same service, so unpublishing
drops the pin rather than leaving a private workflow carrying one.

Two paths are narrowed so a pin can hold. A save wrote the whole row back,
so a publish landing while a save was in flight was silently rolled back,
and a request body could set the publish columns itself; saves now write
only name, description and content. Creating a workflow clears the publish
columns for the same reason.

Nothing reads the pinned copy yet: every workflow is in the following
state it is in today, and nothing on screen changes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../workflow/WorkflowPublishService.scala     | 156 ++++++++
 .../user/workflow/WorkflowResource.scala      |  96 ++++-
 .../user/workflow/WorkflowPublishSpec.scala   | 377 ++++++++++++++++++
 3 files changed, 620 insertions(+), 9 deletions(-)
 create mode 100644 amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
 create mode 100644 amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala

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..aabd5b9fdc0
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishService.scala
@@ -0,0 +1,156 @@
+/*
+ * 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.tables.daos.WorkflowDao
+import org.apache.texera.dao.jooq.generated.tables.pojos.Workflow
+import org.jooq.DSLContext
+
+import scala.util.Try
+import javax.ws.rs.NotFoundException
+
+/**
+  * Version pinning for public workflows.
+  *
+  * A public workflow follows the author's latest content, 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 `workflow.content` until they pin again.
+  *
+  * `is_public` stays the on/off switch; `published_content` is the pin, NULL while following.
+  *
+  * 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 --
+  * which the public read paths take up next.
+  */
+object WorkflowPublishService extends LazyLogging {
+
+  private def context: DSLContext = SqlServer.getInstance().createDSLContext()
+
+  /**
+    * @param hasUnpublishedChanges true when a pin is holding edits back, i.e. pinning again would
+    *                              publish them. 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"))
+
+  /**
+    * Makes the workflow public, following the author's latest. A pin from a previous publication is
+    * not restored: coming back should not silently put old public content on show again.
+    */
+  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 content as the public copy, and turns publishing on. */
+  private def writePin(workflow: Workflow, content: String): Unit =
+    context
+      .update(WORKFLOW)
+      .set(WORKFLOW.IS_PUBLIC, java.lang.Boolean.TRUE)
+      .set(WORKFLOW.PUBLISHED_CONTENT, content)
+      .where(WORKFLOW.WID.eq(workflow.getWid))
+      .execute()
+
+  /**
+    * Clears the pinned copy in one statement, optionally unpublishing too: the CHECK constraint
+    * holds only while the copy and `is_public` move together.
+    *
+    * @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])
+    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 = {
+    val workflow = requireWorkflow(wid)
+    writePin(workflow, workflow.getContent)
+    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 = {
+    clearPin(wid, alsoUnpublish = true)
+    logger.info(s"Workflow $wid unpublished")
+  }
+
+  /** Whether a version is pinned, and whether it is holding edits back. */
+  def statusOf(wid: Integer): PublishStatus = statusOf(requireWorkflow(wid))
+
+  def statusOf(workflow: Workflow): PublishStatus =
+    PublishStatus(
+      isPublished = workflow.getIsPublic,
+      isPinned = workflow.getPublishedContent != null,
+      // Literally "what the public sees is not what you have". Compared on values rather than on a
+      // version id, so that an edit and its undo report nothing held back.
+      hasUnpublishedChanges = workflow.getPublishedContent != null &&
+        !sameContent(workflow.getPublishedContent, workflow.getContent)
+    )
+}
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 e7fecc06c07..72d09654367 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
@@ -93,6 +93,12 @@ object WorkflowResource {
   }
 
   private def insertWorkflow(workflow: Workflow, user: User): Unit = {
+    // A workflow is born with nothing pinned, whether or not the caller asked for a public one:
+    // clearing the columns here is what stops a request body from seeding a public copy of its own.
+    workflow.setPublishedVersionId(null)
+    workflow.setPublishedContent(null)
+    workflow.setPublishedName(null)
+    workflow.setPublishedDescription(null)
     workflowDao.insert(workflow)
     workflowOfUserDao.insert(new WorkflowOfUser(user.getUid, workflow.getWid))
     workflowUserAccessDao.insert(
@@ -104,6 +110,22 @@ object WorkflowResource {
     )
   }
 
+  /**
+    * Writes only the columns a save is allowed to change. Going through the DAO would write the
+    * publish columns from the request body, or restore them from a stale read -- rolling back a pin
+    * that landed in between, or leaving a private workflow carrying one, which the constraint
+    * refuses outright.
+    */
+  private def updateEditableFields(workflow: Workflow): Unit = {
+    context
+      .update(WORKFLOW)
+      .set(WORKFLOW.NAME, workflow.getName)
+      .set(WORKFLOW.DESCRIPTION, workflow.getDescription)
+      .set(WORKFLOW.CONTENT, workflow.getContent)
+      .where(WORKFLOW.WID.eq(workflow.getWid))
+      .execute()
+  }
+
   private def workflowOfUserExists(wid: Integer, uid: Integer): Boolean = {
     workflowOfUserDao.existsById(
       context
@@ -172,9 +194,11 @@ object WorkflowResource {
         user.getUid
       )
     ) {
+      // Same reason as updateEditableFields: a read-modify-write through the DAO would rewrite the
+      // publish columns from a stale read.
       val userWorkflow = workflowDao.fetchOneByWid(wid)
       updateFunction(userWorkflow)
-      workflowDao.update(userWorkflow)
+      updateEditableFields(userWorkflow)
     } else {
       throw new ForbiddenException("No sufficient access privilege.")
     }
@@ -460,7 +484,7 @@ class WorkflowResource extends LazyLogging {
 
     if (workflowOfUserExists(workflow.getWid, user.getUid)) {
       WorkflowVersionResource.insertVersion(workflow, insertingNewWorkflow = false)
-      workflowDao.update(workflow)
+      updateEditableFields(workflow)
     } else {
       if (!WorkflowAccessResource.hasReadAccess(workflow.getWid, user.getUid)) {
         // Check if this workflow exists in the database
@@ -477,7 +501,7 @@ class WorkflowResource extends LazyLogging {
       } else if (WorkflowAccessResource.hasWriteAccess(workflow.getWid, user.getUid)) {
         WorkflowVersionResource.insertVersion(workflow, insertingNewWorkflow = false)
         // not owner but has write access
-        workflowDao.update(workflow)
+        updateEditableFields(workflow)
       } else {
         // not owner and no write access -> rejected
         throw new ForbiddenException("No sufficient access privilege.")
@@ -720,9 +744,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
@@ -732,9 +754,65 @@ 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.
+    if (!WorkflowAccessResource.hasWriteAccess(wid, user.getUid)) {
+      throw new ForbiddenException(s"You do not have permission to modify workflow $wid")
+    }
+    WorkflowPublishService.statusOf(wid)
+  }
+
+  /** The pin endpoints share one guard: writable by this user, and public in the first place. */
+  private def requirePublishable(wid: Integer, user: SessionUser): Unit = {
+    if (!WorkflowAccessResource.hasWriteAccess(wid, user.getUid)) {
+      throw new ForbiddenException(s"You do not have permission to modify workflow $wid")
+    }
+    if (!WorkflowAccessResource.isPublic(wid)) {
+      throw new BadRequestException(s"Workflow $wid is not published")
+    }
   }
 
   /** Returns the workflow's cover image; 404 if none set. */
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..f5f9126e3be
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowPublishSpec.scala
@@ -0,0 +1,377 @@
+/*
+ * 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_USER_ACCESS
+import org.apache.texera.dao.jooq.generated.enums.{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 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 reads the pinned copy yet, so the assertions are
+  * about what is stored; the read paths that serve it arrive with the code that consults it.
+  */
+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)
+  }
+
+  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 "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 "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 "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 "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 "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 "ignore publish columns supplied by the client on create" in {
+    val workflow = new Workflow()
+    workflow.setName("create_cannot_inject")
+    workflow.setDescription("a workflow")
+    workflow.setContent(publishedContent)
+    workflow.setIsPublic(false)
+    // A client cannot hand us a public copy of its own choosing, in any of its parts.
+    workflow.setPublishedContent("""{"operators":[],"note":"injected"}""")
+    workflow.setPublishedName("injected_name")
+    workflow.setPublishedDescription("injected_description")
+    workflow.setPublishedVersionId(1)
+
+    val wid = workflowResource.createWorkflow(workflow, ownerSession).workflow.getWid
+
+    val stored = workflowDao.fetchOneByWid(wid)
+    stored.getPublishedContent shouldBe null
+    stored.getPublishedName shouldBe null
+    stored.getPublishedDescription shouldBe null
+    stored.getPublishedVersionId 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 "reject pinning and unpinning by a user 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 "reject publishing or pinning a workflow that does not exist" in {
+    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.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 read every column and write them all back. A pin landing in that window was
+    // silently reverted to whatever the save had read. The save statement no longer names the
+    // publish columns at all, so the sequence is harmless.
+    val wid = createWorkflow("save_cannot_roll_back_publish")
+    publishPinned(wid)
+
+    // A save built from a snapshot taken *before* a pin that happens in between.
+    val stale = workflowDao.fetchOneByWid(wid)
+    edit(wid, editedContent)
+    workflowResource.pinLatest(wid, ownerSession)
+
+    stale.setContent("""{"operators":[],"note":"from_a_stale_client"}""")
+    workflowResource.persistWorkflow(stale, ownerSession)
+
+    // The pin stands; only the working copy moved.
+    workflowDao.fetchOneByWid(wid).getPublishedContent shouldBe editedContent
+  }
+
+  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 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
+  }
+}

From c51be4860a0ff1a9e9e2188a281fe3ac674033dc Mon Sep 17 00:00:00 2001
From: Yang Zhang <yangz75@uci.edu>
Date: Sat, 22 Aug 2026 14:54:23 -0700
Subject: [PATCH 3/3] feat(workflow): serve the pinned copy to public viewers

With a version pinned, a workflow has two copies: the author's working
copy and the frozen one on public show. This routes every read that
serves a viewer without granted access through the frozen copy, and
freezes the name and description with the graph.

`WorkflowPublishService.publicCopyOf` returns the three fields as a
group, so a surface cannot pick up the published graph under a title the
author has not published; `WorkflowAccessResource.hasGrantedAccess` is
the seam that decides which copy a caller gets. Granted access -- owner,
shared, project member -- keeps tracking the author's latest, because
sharing is not publishing.

Name and description freeze because they are as public as the graph: if
only the graph froze, a report about a title could be answered by editing
the title while the pinned copy still advertised it.

Routed through it: opening a workflow, the hub's read, Clone, Duplicate,
`/workflow_name`, `/workflow_description` and the size a listing shows.
A workflow that follows the author's latest -- every workflow today --
is served exactly what it is served now.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../workflow/WorkflowAccessResource.scala     |   9 +
 .../workflow/WorkflowPublishService.scala     |  45 +++-
 .../user/workflow/WorkflowResource.scala      | 116 +++++---
 .../user/workflow/WorkflowPublishSpec.scala   | 251 +++++++++++++++++-
 4 files changed, 377 insertions(+), 44 deletions(-)

diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala
index 4fd22f4260c..5bfe2df8d05 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResource.scala
@@ -103,6 +103,15 @@ object WorkflowAccessResource {
     }
   }
 
+  /**
+    * Whether the user was granted access to the workflow itself, rather than merely being able to
+    * read it because it is public. Granted access sees the author's working copy; public access is
+    * held at the pinned copy while one is pinned.
+    */
+  def hasGrantedAccess(wid: Integer, uid: Integer): Boolean = {
+    !getPrivilege(wid, uid).eq(PrivilegeEnum.NONE)
+  }
+
   def isPublic(wid: Integer): Boolean = {
     context
       .select(WORKFLOW.IS_PUBLIC)
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 aabd5b9fdc0..9782c475680 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
@@ -88,12 +88,14 @@ object WorkflowPublishService extends LazyLogging {
     statusOf(wid)
   }
 
-  /** Freezes the author's current content as the public copy, and turns publishing on. */
+  /** Writes the pin. Name and description freeze with the graph, or a pin would leak the working title. */
   private def writePin(workflow: Workflow, content: String): Unit =
     context
       .update(WORKFLOW)
       .set(WORKFLOW.IS_PUBLIC, java.lang.Boolean.TRUE)
       .set(WORKFLOW.PUBLISHED_CONTENT, content)
+      .set(WORKFLOW.PUBLISHED_NAME, workflow.getName)
+      .set(WORKFLOW.PUBLISHED_DESCRIPTION, workflow.getDescription)
       .where(WORKFLOW.WID.eq(workflow.getWid))
       .execute()
 
@@ -107,6 +109,8 @@ object WorkflowPublishService extends LazyLogging {
     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])
     val statement =
       if (alsoUnpublish) cleared.set(WORKFLOW.IS_PUBLIC, java.lang.Boolean.FALSE) else cleared
     statement.where(WORKFLOW.WID.eq(wid)).execute()
@@ -148,9 +152,40 @@ object WorkflowPublishService extends LazyLogging {
     PublishStatus(
       isPublished = workflow.getIsPublic,
       isPinned = workflow.getPublishedContent != null,
-      // Literally "what the public sees is not what you have". Compared on values rather than on a
-      // version id, so that an edit and its undo report nothing held back.
-      hasUnpublishedChanges = workflow.getPublishedContent != null &&
-        !sameContent(workflow.getPublishedContent, workflow.getContent)
+      // 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))
     )
+
+  /** Compared as a tree: a restore can rearrange whitespace, and calling that drift alarms nobody. */
+  private def differs(public: PublicCopy, working: PublicCopy): Boolean =
+    public.name != working.name ||
+      public.description != working.description ||
+      !sameContent(public.content, working.content)
+
+  /** Everything about a workflow that is on public show. */
+  case class PublicCopy(name: String, description: String, content: String)
+
+  /** What every public surface must serve, as a group so no field is the one that gets forgotten. */
+  def publicCopyOf(workflow: Workflow): PublicCopy =
+    if (workflow.getPublishedContent == null) workingCopyOf(workflow)
+    else
+      PublicCopy(
+        workflow.getPublishedName,
+        workflow.getPublishedDescription,
+        workflow.getPublishedContent
+      )
+
+  /** The author's own copy, in the same shape. */
+  private def workingCopyOf(workflow: Workflow): PublicCopy =
+    PublicCopy(workflow.getName, workflow.getDescription, workflow.getContent)
+
+  /** As [[publicCopyOf]], for callers holding only a wid. 404s unless the workflow is public. */
+  def publicCopyOf(wid: Integer): PublicCopy = {
+    val workflow = requireWorkflow(wid)
+    if (!workflow.getIsPublic) {
+      throw new NotFoundException(s"Workflow $wid is not public")
+    }
+    publicCopyOf(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 72d09654367..00d8354a5ed 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
@@ -43,7 +43,7 @@ import org.apache.texera.web.resource.dashboard.hub.HubResource.recordCloneActio
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowAccessResource.hasReadAccess
 import org.apache.texera.web.resource.dashboard.user.workflow.WorkflowResource._
 import org.jooq.impl.DSL.{groupConcatDistinct, 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
@@ -95,10 +95,7 @@ object WorkflowResource {
   private def insertWorkflow(workflow: Workflow, user: User): Unit = {
     // A workflow is born with nothing pinned, whether or not the caller asked for a public one:
     // clearing the columns here is what stops a request body from seeding a public copy of its own.
-    workflow.setPublishedVersionId(null)
-    workflow.setPublishedContent(null)
-    workflow.setPublishedName(null)
-    workflow.setPublishedDescription(null)
+    clearPublishState(workflow)
     workflowDao.insert(workflow)
     workflowOfUserDao.insert(new WorkflowOfUser(user.getUid, workflow.getWid))
     workflowUserAccessDao.insert(
@@ -165,11 +162,18 @@ object WorkflowResource {
 
   case class WorkflowIDs(wids: List[Integer], pid: Option[Integer])
 
+  /** Clears every column that describes a pinned public copy. */
+  private def clearPublishState(workflow: Workflow): Unit = {
+    workflow.setPublishedVersionId(null)
+    workflow.setPublishedContent(null)
+    workflow.setPublishedName(null)
+    workflow.setPublishedDescription(null)
+  }
+
   /**
-    * A workflow POJO for the copy-producing paths (clone, duplicate, restore-a-version).
-    *
-    * Built with setters rather than the positional constructor, so that adding a column cannot
-    * silently shift a null into the wrong field -- as adding the published-copy columns would.
+    * A workflow POJO for the copy-producing paths (clone, duplicate, restore-a-version). Copies start
+    * unpublished, and setters rather than the positional constructor keep a new column from silently
+    * shifting a null into the wrong field.
     */
   def newUnpublishedWorkflow(name: String, description: String, content: String): Workflow = {
     val workflow = new Workflow()
@@ -177,9 +181,49 @@ object WorkflowResource {
     workflow.setDescription(description)
     workflow.setContent(content)
     workflow.setIsPublic(false)
+    clearPublishState(workflow)
     workflow
   }
 
+  /**
+    * The copy a viewer may see, all three fields at once: granted access (owner, shared, project
+    * member) sees the author's working copy; everyone else is here because the workflow is public,
+    * and gets the public copy. Taken as a group so a copy cannot end up carrying the published graph
+    * under a title the author has not published.
+    */
+  private def copyVisibleTo(workflow: Workflow, uid: Integer): WorkflowPublishService.PublicCopy =
+    if (uid != null && WorkflowAccessResource.hasGrantedAccess(workflow.getWid, uid)) {
+      WorkflowPublishService.PublicCopy(
+        workflow.getName,
+        workflow.getDescription,
+        workflow.getContent
+      )
+    } else {
+      WorkflowPublishService.publicCopyOf(workflow)
+    }
+
+  /** The content half of [[copyVisibleTo]], for the paths that carry their own name and description. */
+  def contentVisibleTo(workflow: Workflow, uid: Integer): String =
+    copyVisibleTo(workflow, uid).content
+
+  /**
+    * One column as the public sees it: the frozen value while a pin is in place, the live one
+    * otherwise. Keyed on the pin rather than on the frozen value being non-null, so a pinned
+    * workflow can never fall through to what the author is editing.
+    */
+  private def publicField(
+      wid: Integer,
+      frozen: TableField[_, String],
+      live: TableField[_, String]
+  ) =
+    Option(
+      context
+        .select(WORKFLOW.PUBLISHED_CONTENT, frozen, live)
+        .from(WORKFLOW)
+        .where(WORKFLOW.WID.eq(wid))
+        .fetchOne()
+    ).map(row => if (row.value1() != null) row.value2() else row.value3()).orNull
+
   private def updateWorkflowField(
       workflow: Workflow,
       sessionUser: SessionUser,
@@ -452,7 +496,9 @@ class WorkflowResource extends LazyLogging {
         workflow.getName,
         workflow.getDescription,
         workflow.getWid,
-        workflow.getContent,
+        // A user who only reaches this workflow because it is public gets the pinned version, not
+        // the author's in-progress edits.
+        contentVisibleTo(workflow, user.getUid),
         workflow.getCreationTime,
         workflow.getLastModifiedTime,
         workflow.getIsPublic,
@@ -542,11 +588,14 @@ class WorkflowResource extends LazyLogging {
       context.transaction { txConfig =>
         for (wid <- workflowIDs.wids) {
           val oldWorkflow: Workflow = workflowDao.fetchOneByWid(wid)
+          // Reached only because it is public? Then the copy is of the published version, title and
+          // description included.
+          val source = copyVisibleTo(oldWorkflow, user.getUid)
           val newWorkflow = createWorkflow(
             newUnpublishedWorkflow(
-              oldWorkflow.getName + "_copy",
-              oldWorkflow.getDescription,
-              assignNewOperatorIds(oldWorkflow.getContent)
+              source.name + "_copy",
+              source.description,
+              assignNewOperatorIds(source.content)
             ),
             sessionUser
           )
@@ -588,11 +637,14 @@ class WorkflowResource extends LazyLogging {
       throw new ForbiddenException("No sufficient access privilege.")
     }
     val oldWorkflow: Workflow = workflowDao.fetchOneByWid(wid)
+    // The hub shows the public copy, so Clone copies that -- for the author too, who already has
+    // their latest in the editor. For a private workflow this is the author's own copy.
+    val source = WorkflowPublishService.publicCopyOf(oldWorkflow)
     val newWorkflow: DashboardWorkflow = createWorkflow(
       newUnpublishedWorkflow(
-        oldWorkflow.getName + "_clone",
-        oldWorkflow.getDescription,
-        assignNewOperatorIds(oldWorkflow.getContent)
+        source.name + "_clone",
+        source.description,
+        assignNewOperatorIds(source.content)
       ),
       sessionUser
     )
@@ -910,15 +962,10 @@ class WorkflowResource extends LazyLogging {
   @GET
   @Path("/workflow_name")
   def getWorkflowName(@QueryParam("wid") wid: Integer): String = {
-    context
-      .select(
-        WORKFLOW.NAME
-      )
-      .from(WORKFLOW)
-      .where(WORKFLOW.WID.eq(wid))
-      .fetchOneInto(classOf[String])
+    publicField(wid, WORKFLOW.PUBLISHED_NAME, WORKFLOW.NAME)
   }
 
+  /** The hub's public view of a workflow: the pinned version if one is pinned, the latest if not. */
   @GET
   @Path("/publicised/{wid}")
   def retrievePublicWorkflow(
@@ -929,11 +976,14 @@ class WorkflowResource extends LazyLogging {
       .where(WORKFLOW.WID.eq(wid))
       .and(WORKFLOW.IS_PUBLIC.isTrue)
       .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]))
     WorkflowWithPrivilege(
-      workflow.getName,
-      workflow.getDescription,
+      publicCopy.name,
+      publicCopy.description,
       workflow.getWid,
-      workflow.getContent,
+      publicCopy.content,
       workflow.getCreationTime,
       workflow.getLastModifiedTime,
       workflow.getIsPublic,
@@ -944,13 +994,7 @@ class WorkflowResource extends LazyLogging {
   @GET
   @Path("/workflow_description")
   def getWorkflowDescription(@QueryParam("wid") wid: Integer): String = {
-    context
-      .select(
-        WORKFLOW.DESCRIPTION
-      )
-      .from(WORKFLOW)
-      .where(WORKFLOW.WID.eq(wid))
-      .fetchOneInto(classOf[String])
+    publicField(wid, WORKFLOW.PUBLISHED_DESCRIPTION, WORKFLOW.DESCRIPTION)
   }
 
   //TODO Get size from database
@@ -966,7 +1010,11 @@ class WorkflowResource extends LazyLogging {
         .fetch()
         .asScala
         .foreach { wf =>
-          result.put(wf.getWid, wf.getContent.length)
+          // Sized by the copy on show, so the number does not move when the author edits privately.
+          val content =
+            if (wf.getIsPublic && wf.getPublishedContent != null) wf.getPublishedContent
+            else wf.getContent
+          result.put(wf.getWid, content.length)
         }
     }
     result
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 f5f9126e3be..8576adc7b87 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,19 +29,21 @@ 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.user.workflow.WorkflowResource.WorkflowIDs
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import java.lang.reflect.Proxy
 import java.time.OffsetDateTime
+import java.util
+import javax.servlet.http.HttpServletRequest
 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 reads the pinned copy yet, so the assertions are
-  * about what is stored; the read paths that serve it arrive with the code that consults it.
+  * 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.
   */
 class WorkflowPublishSpec
     extends AnyFlatSpec
@@ -112,6 +114,25 @@ class WorkflowPublishSpec
     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)
+  }
+
+  /** Clone records the caller's IP, and that is the only thing it wants from the request. */
+  private def fakeRequest(): HttpServletRequest =
+    Proxy
+      .newProxyInstance(
+        classOf[HttpServletRequest].getClassLoader,
+        Array[Class[_]](classOf[HttpServletRequest]),
+        (_: Any, method: java.lang.reflect.Method, _: Array[AnyRef]) =>
+          if (method.getName == "getRemoteAddr") "127.0.0.1" else null
+      )
+      .asInstanceOf[HttpServletRequest]
+
   private def statusOf(wid: Integer): WorkflowPublishService.PublishStatus =
     workflowResource.getPublishStatus(wid, ownerSession)
 
@@ -374,4 +395,224 @@ class WorkflowPublishSpec
     stored.getIsPublic shouldBe true
     stored.getPublishedContent shouldBe publishedContent
   }
+
+  behavior of "name and description"
+
+  it should "freeze the name and description alongside the content" in {
+    // Malicious text in a description is just as public as the graph, so editing it must not reach
+    // the public view either -- otherwise a report can be answered by rewording rather than fixing.
+    val wid = createWorkflow("freeze_metadata")
+    publishPinned(wid)
+
+    relabel(wid, "renamed_after_publishing", "rewritten after publishing")
+
+    val publicView = workflowResource.retrievePublicWorkflow(wid)
+    publicView.name shouldBe "freeze_metadata"
+    publicView.description shouldBe "a workflow"
+    workflowResource.getWorkflowName(wid) shouldBe "freeze_metadata"
+    workflowResource.getWorkflowDescription(wid) shouldBe "a workflow"
+  }
+
+  it should "count a description edit as an unpublished change" in {
+    val wid = createWorkflow("description_counts")
+    publishPinned(wid)
+    statusOf(wid).hasUnpublishedChanges shouldBe false
+
+    relabel(wid, "description_counts", "rewritten after publishing")
+
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+    workflowResource.pinLatest(wid, ownerSession)
+    workflowResource.retrievePublicWorkflow(wid).description shouldBe "rewritten after publishing"
+  }
+
+  behavior of "public read paths"
+
+  it should "show a public viewer the author's latest while nothing is pinned" in {
+    // Following follows everything a pin would freeze, not just the graph: name and description are
+    // on public show too, so a viewer must see the live ones or the two halves would disagree.
+    val wid = createWorkflow("following_serves_latest_to_strangers")
+    workflowResource.makePublic(wid, ownerSession)
+    edit(wid, editedContent)
+    relabel(wid, "renamed_while_following", "described_while_following")
+
+    workflowResource.retrieveWorkflow(wid, strangerSession).content shouldBe editedContent
+    workflowResource.getWorkflowName(wid) shouldBe "renamed_while_following"
+    workflowResource.getWorkflowDescription(wid) shouldBe "described_while_following"
+
+    val publicView = workflowResource.retrievePublicWorkflow(wid)
+    publicView.content shouldBe editedContent
+    publicView.name shouldBe "renamed_while_following"
+    publicView.description shouldBe "described_while_following"
+  }
+
+  it should "serve the published version to a user without granted access" in {
+    val wid = createWorkflow("read_serves_published")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    // A stranger reaches this workflow only because it is public.
+    workflowResource.retrieveWorkflow(wid, strangerSession).content shouldBe publishedContent
+    // The author keeps seeing their own working copy.
+    workflowResource.retrieveWorkflow(wid, ownerSession).content shouldBe editedContent
+  }
+
+  it should "serve the working copy to a collaborator with granted read access" in {
+    val wid = createWorkflow("collaborator_sees_working_copy")
+    publishPinned(wid)
+    edit(wid, editedContent)
+    grantAccess(wid, PrivilegeEnum.READ)
+
+    try {
+      // Sharing is not publishing: a collaborator tracks the author's latest content, live.
+      workflowResource.retrieveWorkflow(wid, strangerSession).content shouldBe editedContent
+    } finally revokeAccess(wid)
+  }
+
+  it should "keep serving a collaborator the latest content as the author keeps editing" in {
+    val wid = createWorkflow("collaborator_tracks_latest")
+    publishPinned(wid)
+    grantAccess(wid, PrivilegeEnum.READ)
+
+    try {
+      val later = """{"operators":[],"note":"later_still"}"""
+      edit(wid, editedContent)
+      workflowResource.retrieveWorkflow(wid, strangerSession).content shouldBe editedContent
+      edit(wid, later)
+      workflowResource.retrieveWorkflow(wid, strangerSession).content shouldBe later
+      // ...while the public copy stayed put throughout.
+      workflowResource.retrievePublicWorkflow(wid).content shouldBe publishedContent
+    } finally revokeAccess(wid)
+  }
+
+  it should "not tell a public viewer that the author has unpublished edits" in {
+    val wid = createWorkflow("draft_state_is_private")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    a[ForbiddenException] should be thrownBy workflowResource.getPublishStatus(wid, strangerSession)
+    statusOf(wid).hasUnpublishedChanges shouldBe true
+  }
+
+  it should "refuse to hand out a public copy of a workflow that is not public" in {
+    // The guard viewers without granted access rely on: no route to a private workflow's content
+    // may fall through to the public copy just because the caller asked for it by wid.
+    val wid = createWorkflow("public_copy_requires_public")
+    a[NotFoundException] should be thrownBy WorkflowPublishService.publicCopyOf(wid)
+  }
+
+  it should "clone the published version, not the author's latest" in {
+    val wid = createWorkflow("clone_takes_published")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    val clonedWid = workflowResource.cloneWorkflow(wid, strangerSession, fakeRequest())
+    val cloned = workflowDao.fetchOneByWid(clonedWid)
+
+    cloned.getContent shouldBe publishedContent
+    // A copy has never been reviewed, so it starts private.
+    cloned.getIsPublic shouldBe false
+  }
+
+  it should "clone the published version for the author too" in {
+    // The hub shows the pinned version, so its Clone button copies that even for the author, whose
+    // working copy has moved on -- cloning something other than what is on the screen would be the
+    // surprise, and their latest is already open in the editor.
+    val wid = createWorkflow("clone_takes_published_for_author")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    val clonedWid = workflowResource.cloneWorkflow(wid, ownerSession, fakeRequest())
+
+    workflowDao.fetchOneByWid(clonedWid).getContent shouldBe publishedContent
+  }
+
+  it should "clone the published name and description, not the edited ones" in {
+    val wid = createWorkflow("clone_takes_published_metadata")
+    publishPinned(wid)
+    relabel(wid, "renamed_after_publishing", "described_after_publishing")
+
+    val cloned =
+      workflowDao.fetchOneByWid(workflowResource.cloneWorkflow(wid, ownerSession, fakeRequest()))
+
+    cloned.getName shouldBe "clone_takes_published_metadata_clone"
+    cloned.getDescription shouldBe "a workflow"
+  }
+
+  it should "still clone the working copy of a workflow that is not public" in {
+    val wid = createWorkflow("clone_private_takes_working_copy")
+    edit(wid, editedContent)
+
+    val clonedWid = workflowResource.cloneWorkflow(wid, ownerSession, fakeRequest())
+
+    workflowDao.fetchOneByWid(clonedWid).getContent shouldBe editedContent
+  }
+
+  it should "duplicate the published version for a user without granted access" in {
+    // Title and description too, not just the graph: a copy carrying the published canvas under the
+    // author's unpublished title would publish the very rename the pin is holding back.
+    val wid = createWorkflow("duplicate_takes_published")
+    publishPinned(wid)
+    edit(wid, editedContent)
+    relabel(wid, "duplicate_unpublished_name", "unpublished description")
+
+    val duplicated =
+      workflowResource.duplicateWorkflow(WorkflowIDs(List(wid), None), strangerSession)
+
+    duplicated should have size 1
+    val copy = workflowDao.fetchOneByWid(duplicated.head.workflow.getWid)
+    copy.getContent shouldBe publishedContent
+    copy.getName shouldBe "duplicate_takes_published_copy"
+    copy.getDescription should not be "unpublished description"
+  }
+
+  it should "duplicate the owner's own working copy for the owner" in {
+    val wid = createWorkflow("owner_duplicates_working_copy")
+    publishPinned(wid)
+    edit(wid, editedContent)
+
+    val duplicated = workflowResource.duplicateWorkflow(WorkflowIDs(List(wid), None), ownerSession)
+    workflowDao.fetchOneByWid(duplicated.head.workflow.getWid).getContent shouldBe editedContent
+  }
+
+  it should "start a copy of a published workflow with no publish state of its own" in {
+    val wid = createWorkflow("copy_starts_clean")
+    publishPinned(wid)
+
+    val copy = workflowDao.fetchOneByWid(
+      workflowResource
+        .duplicateWorkflow(WorkflowIDs(List(wid), None), ownerSession)
+        .head
+        .workflow
+        .getWid
+    )
+
+    copy.getIsPublic shouldBe false
+    copy.getPublishedContent shouldBe null
+    copy.getPublishedName shouldBe null
+    copy.getPublishedDescription shouldBe null
+    copy.getPublishedVersionId shouldBe null
+  }
+
+  it should "size a workflow by the copy the caller can see" in {
+    // Listings show a size next to every card, so it has to describe the copy that card opens: the
+    // pinned one while a pin is in place, the author's latest otherwise, and a private workflow's
+    // own content.
+    val priv = createWorkflow("private_size_uses_content")
+    edit(priv, editedContent)
+    workflowResource.getSize(util.Arrays.asList(priv)).get(priv) shouldBe editedContent.length
+
+    val following = createWorkflow("size_follows_latest")
+    workflowResource.makePublic(following, ownerSession)
+    edit(following, editedContent + "     ")
+    workflowResource
+      .getSize(util.Arrays.asList(following))
+      .get(following) shouldBe editedContent.length + 5
+
+    val pinned = createWorkflow("size_uses_published")
+    publishPinned(pinned)
+    edit(pinned, editedContent + "                    ")
+    workflowResource
+      .getSize(util.Arrays.asList(pinned))
+      .get(pinned) shouldBe publishedContent.length
+  }
 }