fix(forks): route new workflows through one row builder and harden lineage locking (#8406)

* fix(forks): route new workflows through one row builder and harden lineage locking

* fix(workflows): read the fork-sync policy inside each create transaction

Author: waleedlatif1

apps/docs/content/docs/platform/enterprise/forks.mdx

@@ -29,7 +29,7 @@ On Sim Cloud, your organization may also need the feature turned on for your acc
 
 ### 1. Open Forks
 
-Go to **Settings → Organization → Workspace forks** in the workspace you want to fork from (or manage).
+Go to **Settings → Workspace → Workspace forks** in the workspace you want to fork from (or manage).
 
 <Image src="/static/enterprise/forks-list.png" alt="Workspace Forks settings page showing Parent and Forks sections with Docs, See activity, and Create fork actions" width={900} height={369} />
 
@@ -62,7 +62,7 @@ Click **Fork**. The child workspace is created immediately. Deployed workflows l
 
 ### 3. Open the parent edge (from the child)
 
-Open the **child** workspace → **Settings → Organization → Workspace forks**. On the **Parent** row, open the menu and choose **Edit mappings**.
+Open the **child** workspace → **Settings → Workspace → Workspace forks**. On the **Parent** row, open the menu and choose **Edit mappings**.
 
 Child rows (when you are on the parent) only offer **Open workspace** and **Disconnect** — mapping and sync are owned by the child configuring how it relates to its parent.
 
@@ -136,14 +136,14 @@ Above the list, **Sync new workflows by default** decides where a **newly create
 
 | Setting | A new workflow… |
 |---------|-----------------|
-| **On** (default) | joins fork sync — it arrives checked and syncs as soon as you deploy it |
-| **Off** | starts outside fork sync — it arrives unchecked and only syncs after you check it |
+| **Sync** (default) | joins fork sync — it arrives checked and syncs as soon as you deploy it |
+| **Don't sync** | starts outside fork sync — it arrives unchecked and only syncs after you check it |
 
 Three things to know:
 
-- **It applies to the whole fork lineage.** The toggle writes every workspace in the lineage — the root, every ancestor, every descendant — so a parent and its forks can never disagree about what "new" means. Any workspace admin in the lineage can change it, and each member gets its own audit entry naming the workspace the change came from. A new fork inherits the value at creation.
+- **It applies to the whole fork lineage.** The toggle writes every workspace in the lineage — the root, every ancestor, every descendant — so a parent and its forks can never disagree about what "new" means. Any workspace admin in the lineage can change it, and each workspace whose value changes gets its own audit entry naming the workspace the change came from. A new fork inherits the value at creation.
 - **It is forward-only.** Flipping it never moves an existing workflow in or out of sync. The checkbox list above stays the record of what syncs.
-- **"New" means genuinely new.** Creating, duplicating, or importing a workflow takes this setting, as does the blank starter workflow a fork gets when there is nothing to copy. A workflow that arrives as a **copy** — from a fork, or from a push or pull — inherits its source's own checkbox instead, so a workflow you deliberately synced never lands unsynced in the child.
+- **"New" means genuinely new.** Creating, duplicating, or importing a workflow takes this setting, as does the blank starter workflow a fork gets when there is nothing to copy. A workflow that arrives as a **copy** — from a fork, or from a push or pull — ignores this setting. Only synced workflows are copied, and they always arrive synced, so a workflow you deliberately synced never lands unsynced on the other side.
 
 **Example:** a template workspace turns this off so every scratch workflow the team creates stays local, then checks only the handful meant to reach the forks.
 
@@ -399,7 +399,7 @@ Schedules, webhooks, and triggers are not live in the child until you **deploy**
   { question: "Why is Sync greyed out?", answer: "Usually a blocking reference, an unmapped credential or secret, or a required dependent field (label, channel, document, …) still empty. Open Blocking sync and the mapping sections — each row explains what to fix. Sync also stays disabled while details are loading or if loading failed (reload the page)." },
   { question: "Is sync a merge?", answer: "No. Deploy is like a commit; sync is a force push or force pull of deployed workflows onto the target. Use Rollback only for the last sync into a workspace, and remember copied resources may remain." },
   { question: "Who can disconnect a fork I cannot open?", answer: "Any admin on your side of the edge. Disconnect does not require access to the other workspace — so you are not stuck if the other side lost membership." },
-  { question: "I deployed a new workflow and sync ignored it. Why?", answer: "Sync new workflows by default is off for this fork lineage, so the workflow was created outside fork sync. Open Settings → Organization → Workspace forks and check it under Synced workflows. Turning the toggle back on only affects workflows created after that — it never moves an existing one." },
+  { question: "I deployed a new workflow and sync ignored it. Why?", answer: "Sync new workflows by default is off for this fork lineage, so the workflow was created outside fork sync. Open Settings → Workspace → Workspace forks and check it under Synced workflows. Turning the toggle back on only affects workflows created after that — it never moves an existing one." },
   { question: "Does turning Sync new workflows by default off stop my current syncs?", answer: "No. It is forward-only and never rewrites an existing workflow's checkbox, so everything already synced keeps syncing. It also applies to every workspace in the fork lineage, not just the one you changed it from." }
 ]} />
 
@@ -413,4 +413,4 @@ Self-hosted deployments turn Forks on with an environment variable instead of th
 |----------|-------------|
 | `FORKING_ENABLED`, `NEXT_PUBLIC_FORKING_ENABLED` | Enables workspace forking when billing is not used as the entitlement gate |
 
-Once enabled, use the same **Settings → Organization → Workspace forks** UI as Sim Cloud. Only workspace admins can manage forks.
+Once enabled, use the same **Settings → Workspace → Workspace forks** UI as Sim Cloud. Only workspace admins can manage forks.

apps/sim/app/api/superuser/import-workflow/route.ts

@@ -12,13 +12,13 @@ import { loadCopilotChatMessages } from '@/lib/mothership/chat/lifecycle'
 import { appendCopilotChatMessages } from '@/lib/mothership/chat/messages-store'
 import { verifyEffectiveSuperUser } from '@/lib/permissions/super-user'
 import { parseWorkflowJson } from '@/lib/workflows/operations/import-export'
+import { buildNewWorkflowRow } from '@/lib/workflows/persistence/new-workflow-row'
 import {
   loadWorkflowFromNormalizedTables,
   saveWorkflowToNormalizedTables,
 } from '@/lib/workflows/persistence/utils'
 import { sanitizeForExport } from '@/lib/workflows/sanitization/json-sanitizer'
 import { deduplicateWorkflowName } from '@/lib/workflows/utils'
-import { resolveForkSyncExclusionForNewWorkflow } from '@/ee/workspace-forking/lib/sync-default'
 
 const logger = createLogger('SuperUserImportWorkflow')
 
@@ -130,30 +130,23 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
 
     // Create new workflow record
     const newWorkflowId = generateId()
-    const now = new Date()
     const dedupedName = await deduplicateWorkflowName(
       `[Debug Import] ${sourceWorkflow.name}`,
       targetWorkspaceId,
       null
     )
 
-    await db.insert(workflow).values({
-      id: newWorkflowId,
-      userId: session.user.id,
-      workspaceId: targetWorkspaceId,
-      folderId: null,
-      name: dedupedName,
-      description: sourceWorkflow.description,
-      lastSynced: now,
-      createdAt: now,
-      updatedAt: now,
-      isDeployed: false, // Never copy deployment status
-      runCount: 0,
-      variables: sourceWorkflow.variables || {},
-      // An imported workflow is a NEW workflow in the target workspace, so it takes that
-      // workspace's fork-sync policy rather than the column default.
-      forkSyncExcluded: await resolveForkSyncExclusionForNewWorkflow(db, targetWorkspaceId),
-    })
+    await db.insert(workflow).values(
+      await buildNewWorkflowRow(db, {
+        id: newWorkflowId,
+        userId: session.user.id,
+        workspaceId: targetWorkspaceId,
+        folderId: null,
+        name: dedupedName,
+        description: sourceWorkflow.description,
+        variables: sourceWorkflow.variables || {},
+      })
+    )
 
     // Save using existing persistence logic
     const saveResult = await saveWorkflowToNormalizedTables(newWorkflowId, importedData, {

apps/sim/app/api/v1/admin/workflows/import/route.ts

@@ -30,6 +30,7 @@ import { adminV1ImportWorkflowContract } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { parseWorkflowJson } from '@/lib/workflows/operations/import-export'
+import { buildNewWorkflowRow } from '@/lib/workflows/persistence/new-workflow-row'
 import { prepareWorkflowStateForPersistence } from '@/lib/workflows/persistence/prepare-state'
 import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils'
 import { deduplicateWorkflowName } from '@/lib/workflows/utils'
@@ -41,7 +42,6 @@ import {
   notFoundResponse,
 } from '@/app/api/v1/admin/responses'
 import { extractWorkflowMetadata, type WorkflowImportRequest } from '@/app/api/v1/admin/types'
-import { resolveForkSyncExclusionForNewWorkflow } from '@/ee/workspace-forking/lib/sync-default'
 
 const logger = createLogger('AdminWorkflowImportAPI')
 
@@ -113,27 +113,18 @@ export const POST = withRouteHandler(
       )
 
       const workflowId = generateId()
-      const now = new Date()
       const dedupedName = await deduplicateWorkflowName(workflowName, workspaceId, folderId || null)
 
-      await db.insert(workflow).values({
-        id: workflowId,
-        userId: workspaceData.ownerId,
-        workspaceId,
-        folderId: folderId || null,
-        name: dedupedName,
-        description: workflowDescription,
-        lastSynced: now,
-        createdAt: now,
-        updatedAt: now,
-        isDeployed: false,
-        runCount: 0,
-        variables: {},
-        // An imported workflow is a NEW workflow in this workspace, so it takes the
-        // workspace's fork-sync policy. Without this it lands on the column default and
-        // silently joins fork sync in a workspace that opted out.
-        forkSyncExcluded: await resolveForkSyncExclusionForNewWorkflow(db, workspaceId),
-      })
+      await db.insert(workflow).values(
+        await buildNewWorkflowRow(db, {
+          id: workflowId,
+          userId: workspaceData.ownerId,
+          workspaceId,
+          folderId: folderId || null,
+          name: dedupedName,
+          description: workflowDescription,
+        })
+      )
 
       /**
        * Same normalization the editor and the v1 import API run, via the one

apps/sim/app/api/v1/admin/workspaces/[id]/import/route.ts

@@ -45,6 +45,7 @@ import {
   extractWorkflowsFromZip,
   parseWorkflowJson,
 } from '@/lib/workflows/operations/import-export'
+import { buildNewWorkflowRow } from '@/lib/workflows/persistence/new-workflow-row'
 import { prepareWorkflowStateForPersistence } from '@/lib/workflows/persistence/prepare-state'
 import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils'
 import { deduplicateWorkflowName } from '@/lib/workflows/utils'
@@ -62,7 +63,6 @@ import type {
   WorkspaceImportRequest,
   WorkspaceImportResponse,
 } from '@/app/api/v1/admin/types'
-import { resolveForkSyncExclusionForNewWorkflow } from '@/ee/workspace-forking/lib/sync-default'
 
 const logger = createLogger('AdminWorkspaceImportAPI')
 
@@ -348,27 +348,18 @@ async function importSingleWorkflow(
     }
 
     const workflowId = generateId()
-    const now = new Date()
     const dedupedName = await deduplicateWorkflowName(workflowName, workspaceId, targetFolderId)
 
-    await db.insert(workflow).values({
-      id: workflowId,
-      userId: ownerId,
-      workspaceId,
-      folderId: targetFolderId,
-      name: dedupedName,
-      description: workflowData.metadata?.description || 'Imported via Admin API',
-      lastSynced: now,
-      createdAt: now,
-      updatedAt: now,
-      isDeployed: false,
-      runCount: 0,
-      variables: {},
-      // An imported workflow is a NEW workflow in this workspace, so it takes the
-      // workspace's fork-sync policy. Without this it lands on the column default and
-      // silently joins fork sync in a workspace that opted out.
-      forkSyncExcluded: await resolveForkSyncExclusionForNewWorkflow(db, workspaceId),
-    })
+    await db.insert(workflow).values(
+      await buildNewWorkflowRow(db, {
+        id: workflowId,
+        userId: ownerId,
+        workspaceId,
+        folderId: targetFolderId,
+        name: dedupedName,
+        description: workflowData.metadata?.description || 'Imported via Admin API',
+      })
+    )
 
     /**
      * Same normalization the editor, the v1 import API and the single-workflow

apps/sim/app/api/workspaces/[id]/fork/sync-default/route.ts

@@ -13,10 +13,8 @@ export const PUT = defineInternalJsonRoute({
   auth: internalSessionAuth,
   operation: forkOperations.syncDefault,
   /**
-   * Rated, unlike its sibling fork routes. This is the one that writes workspaces the
-   * caller may not administer, under the feature's coarsest advisory lock, so an admin of
-   * any single lineage member could otherwise loop it and starve fork creation across the
-   * whole lineage.
+   * Rate-limited, unlike sibling fork routes: it writes the whole lineage under the coarsest
+   * fork lock, so looping it could starve fork creation lineage-wide.
    */
   rateLimit: internalRateLimits.user({ bucketName: 'workspace-fork-sync-default' }),
   errorPolicy: internalForkErrorPolicy,

apps/sim/ee/workspace-forking/application/lineage-details.ts

@@ -1,10 +1,10 @@
 import { db } from '@sim/db'
 import { workspace } from '@sim/db/schema'
 import { eq } from 'drizzle-orm'
+import { readForkSyncNewWorkflowsExcluded } from '@/lib/workflows/persistence/new-workflow-row'
 import { getEffectiveWorkspacePermission } from '@/lib/workspaces/permissions/utils'
 import { getForkChildren, getForkParent } from '@/ee/workspace-forking/lib/lineage/lineage'
 import { getUndoableRunForTarget } from '@/ee/workspace-forking/lib/promote/promote-run-store'
-import { resolveForkSyncExclusionForNewWorkflow } from '@/ee/workspace-forking/lib/sync-default'
 
 /**
  * Annotates a lineage node with whether the viewer holds any access to it (explicit
@@ -40,7 +40,7 @@ export const getWorkspaceForkLineageDetails = defineForkUseCase({
       getForkChildren(workspaceId),
       getUndoableRunForTarget(db, workspaceId),
       // Lineage-uniform, so this workspace's own value is the lineage's value.
-      resolveForkSyncExclusionForNewWorkflow(db, workspaceId),
+      readForkSyncNewWorkflowsExcluded(db, workspaceId),
     ])
 
     const [parent, children] = await Promise.all([

apps/sim/ee/workspace-forking/application/operations.ts

@@ -99,12 +99,8 @@ export const forkOperations = {
     oauthScope: 'api:write',
   }),
   /**
-   * Admin on the CALLING workspace is sufficient, and the write then fans out to every
-   * ancestor and descendant, because the default is meaningless unless it is uniform
-   * across a lineage. Flipping it to "sync new workflows" restores the historical
-   * behaviour rather than granting anything new, and it never moves an existing workflow
-   * in or out of sync - so each member records its own audit entry rather than the write
-   * being restricted to one workspace.
+   * Admin on the calling workspace is sufficient; the write fans out to the whole lineage
+   * because the default must be uniform, and it never moves an existing workflow.
    *
    * permission-group-exempt: the new-workflow fork-sync default is workspace configuration governed by the admin role.
    */

apps/sim/ee/workspace-forking/application/recovery-and-mappings.ts

@@ -52,10 +52,8 @@ export const updateWorkspaceForkMappings = defineForkUseCase<
       input.direction === 'push' ? input.otherWorkspaceId : input.workspaceId
     return db.transaction(async (tx) => {
       await setForkLockTimeout(tx)
-      // Rank 4 - see the rank table on `acquireForkLineageLock`. Unlike promote and
-      // rollback this takes no rank-3 target lock: it rewrites only this edge's mapping
-      // rows, never the target's workflows, so nothing contends with a sync into the
-      // target. Skipping a higher rank is not an ordering violation.
+      // Rank 4 - see the rank table on `acquireForkLineageLock`. No target lock: this
+      // rewrites only the edge's mapping rows.
       await acquireForkEdgeLock(tx, edge.childWorkspaceId)
       const [currentEdge] = await tx
         .select({ parentId: workspace.forkedFromWorkspaceId })

apps/sim/ee/workspace-forking/application/revision.ts

@@ -143,11 +143,8 @@ export async function loadForkPreviewRevision(
 /**
  * Locks normalized graph rows as well as workflow metadata, including realtime-only writes.
  *
- * Rank 5 - the heaviest acquirer in the fork module, and the one the rank table on
- * `acquireForkLineageLock` exists for. It takes `FOR UPDATE` on the `workspace` rows, so
- * any caller that also needs the rank-2 lineage lock must take that one FIRST; doing it
- * the other way round deadlocks against `unlinkForkEdge`, which holds the lineage key and
- * then updates the same `workspace` row.
+ * Rank 5 - see the rank table on `acquireForkLineageLock`. Takes `FOR UPDATE` on `workspace`
+ * rows, so a caller needing the rank-2 lineage lock must take it first.
  */
 export async function lockForkRevision(tx: DbTransaction, scope: ForkRevisionScope): Promise<void> {
   const workspaceIds = [
@@ -207,21 +204,11 @@ export async function assertForkSourceVersions(
   sourceWorkspaceId: string,
   expected: ReadonlyMap<string, { id: string; digest: string }>
 ): Promise<void> {
-  // Verify exactly the workflows that were ADMITTED, rather than re-deriving the source
-  // predicate here. Re-deriving it duplicated `listDeployedWorkflows`'s filter, so the day
-  // a caller admitted a different set - "Copy unsynced workflows" admits sync-excluded
-  // workflows - this query returned fewer rows and every such fork failed on a phantom
-  // size mismatch. Keying off `expected` cannot drift from the admitted set by construction.
-  if (expected.size === 0) return
-  const admittedIds = sql.join(
-    [...expected.keys()].map((id) => sql`${id}`),
-    sql`, `
-  )
   const rows = await tx.execute<{ workflowId: string; id: string; digest: string }>(sql`
     SELECT w.id AS "workflowId", d.id, md5(d.state::text) AS digest FROM ${workflow} w
     JOIN ${workflowDeploymentVersion} d ON d.workflow_id = w.id AND d.is_active = true
     WHERE w.workspace_id = ${sourceWorkspaceId} AND w.is_deployed = true
-      AND w.archived_at IS NULL AND w.id IN (${admittedIds})
+      AND w.archived_at IS NULL AND w.fork_sync_excluded = false
   `)
   if (
     rows.length !== expected.size ||