fix(workflows): resume legacy execution snapshots

Author: TheodoreSpeaks

apps/sim/executor/execution/snapshot.test.ts

@@ -54,6 +54,7 @@ describe('ExecutionSnapshot', () => {
     expect(() =>
       ExecutionSnapshot.fromJSON(
         JSON.stringify({
+          version: 1,
           metadata: { ...metadata, principal: { version: 99, principal: {} } },
           workflow: { blocks: [] },
           input: {},
@@ -70,6 +71,7 @@ describe('ExecutionSnapshot', () => {
     expect(() =>
       ExecutionSnapshot.fromJSON(
         JSON.stringify({
+          version: 1,
           metadata: metadataWithoutPrincipal,
           workflow: { blocks: [] },
           input: {},
@@ -79,4 +81,58 @@ describe('ExecutionSnapshot', () => {
       )
     ).toThrow('Execution snapshot metadata is missing its principal')
   })
+
+  it('restores the recorded session user from a legacy pause snapshot', () => {
+    const { principal: _principal, ...legacyMetadata } = metadata
+    const restored = ExecutionSnapshot.fromJSON(
+      JSON.stringify({
+        metadata: { ...legacyMetadata, sessionUserId: 'session-user-1' },
+        workflow: { blocks: [] },
+        input: {},
+        workflowVariables: {},
+        selectedOutputs: [],
+      })
+    )
+
+    expect(restored.metadata.principal).toEqual({
+      kind: 'session',
+      userId: 'session-user-1',
+      sessionId: 'legacy-paused-execution',
+    })
+  })
+
+  it('restores actorless legacy pause snapshots as internal system executions', () => {
+    const { principal: _principal, ...legacyMetadata } = metadata
+    const restored = ExecutionSnapshot.fromJSON(
+      JSON.stringify({
+        metadata: legacyMetadata,
+        workflow: { blocks: [] },
+        input: {},
+        workflowVariables: {},
+        selectedOutputs: [],
+      })
+    )
+
+    expect(restored.metadata.principal).toEqual({
+      kind: 'system',
+      serviceId: 'internal',
+      workspaceId: 'workspace-1',
+      workflowId: 'workflow-1',
+    })
+  })
+
+  it('rejects unsupported execution snapshot versions', () => {
+    expect(() =>
+      ExecutionSnapshot.fromJSON(
+        JSON.stringify({
+          version: 2,
+          metadata,
+          workflow: { blocks: [] },
+          input: {},
+          workflowVariables: {},
+          selectedOutputs: [],
+        })
+      )
+    ).toThrow('Unsupported execution snapshot version 2')
+  })
 })

apps/sim/executor/execution/snapshot.ts

@@ -1,8 +1,43 @@
-import { parsePrincipal, serializePrincipal } from '@sim/auth/principal'
+import {
+  parsePrincipal,
+  serializePrincipal,
+  type WorkflowExecutionPrincipal,
+} from '@sim/auth/principal'
 import { normalizeStringArray } from '@/lib/core/utils/arrays'
 import { normalizeWorkflowVariables } from '@/lib/core/utils/records'
 import type { ExecutionMetadata, SerializableExecutionState } from '@/executor/execution/types'
 
+const EXECUTION_SNAPSHOT_VERSION = 1
+const LEGACY_PAUSE_SESSION_ID = 'legacy-paused-execution'
+
+function requireLegacyMetadataString(
+  metadata: Record<string, unknown>,
+  field: 'workflowId' | 'workspaceId'
+): string {
+  const value = metadata[field]
+  if (typeof value !== 'string' || !value.trim()) {
+    throw new Error(`Legacy execution snapshot metadata ${field} must be a non-empty string`)
+  }
+  return value
+}
+
+/** Restores only identity that the pre-principal snapshot format recorded unambiguously. */
+function parseLegacyPrincipal(metadata: Record<string, unknown>): WorkflowExecutionPrincipal {
+  const workflowId = requireLegacyMetadataString(metadata, 'workflowId')
+  const workspaceId = requireLegacyMetadataString(metadata, 'workspaceId')
+  if (metadata.sessionUserId !== undefined) {
+    if (typeof metadata.sessionUserId !== 'string' || !metadata.sessionUserId.trim()) {
+      throw new Error('Legacy execution snapshot metadata sessionUserId must be a non-empty string')
+    }
+    return {
+      kind: 'session',
+      userId: metadata.sessionUserId,
+      sessionId: LEGACY_PAUSE_SESSION_ID,
+    }
+  }
+  return { kind: 'system', serviceId: 'internal', workspaceId, workflowId }
+}
+
 export class ExecutionSnapshot {
   public readonly metadata: ExecutionMetadata
   public readonly workflow: any
@@ -29,6 +64,7 @@ export class ExecutionSnapshot {
 
   toJSON(): string {
     return JSON.stringify({
+      version: EXECUTION_SNAPSHOT_VERSION,
       metadata: {
         ...this.metadata,
         principal: serializePrincipal(this.metadata.principal),
@@ -51,12 +87,23 @@ export class ExecutionSnapshot {
       throw new Error('Execution snapshot metadata must be an object')
     }
     const serializedMetadata = parsed.metadata as Record<string, unknown>
-    if (serializedMetadata.principal === undefined) {
-      throw new Error('Execution snapshot metadata is missing its principal')
+    let principal: WorkflowExecutionPrincipal
+    if (parsed.version === EXECUTION_SNAPSHOT_VERSION) {
+      if (serializedMetadata.principal === undefined) {
+        throw new Error('Execution snapshot metadata is missing its principal')
+      }
+      principal = parsePrincipal(serializedMetadata.principal)
+    } else if (parsed.version === undefined) {
+      if (serializedMetadata.principal !== undefined) {
+        throw new Error('Unversioned execution snapshots cannot contain a principal')
+      }
+      principal = parseLegacyPrincipal(serializedMetadata)
+    } else {
+      throw new Error(`Unsupported execution snapshot version ${String(parsed.version)}`)
     }
     const metadata = {
       ...serializedMetadata,
-      principal: parsePrincipal(serializedMetadata.principal),
+      principal,
     } as ExecutionMetadata
     return new ExecutionSnapshot(
       metadata,