fix(workflows): refuse stop targets a run cannot reach, and run the E2E self-hosted

- The manual operations refuse a stop block the run cannot reach from its
  entry (an upstream block would let the run finish everything after the
  entry), and look blocks up as own properties.
- The executor fails a run whose stop block is absent from the workflow it
  executes, instead of running everything; this closes the window between
  validation and the executor's own draft load, for every caller.
- The CLI refuses an empty --stop-after rather than dropping it.
- CI: the stop-after E2E gets its own self-hosted app step; the SCIM suite
  asserts PostgreSQL rate-limit storage, so Redis is not added to that app.
  Fixture cleanup waits for run logs to finalize before deleting.

Author: Waleed Latif

.github/workflows/test-build.yml

@@ -125,16 +125,6 @@ jobs:
     runs-on: ${{ (vars.CI_PROVIDER == '' || vars.CI_PROVIDER == 'blacksmith') && 'blacksmith-8vcpu-ubuntu-2404' || 'ubuntu-latest' }}
     timeout-minutes: 20
     services:
-      # Hosted billing admits a workflow run only through a Redis usage reservation.
-      redis:
-        image: redis:8.2-alpine
-        ports:
-          - 6379:6379
-        options: >-
-          --health-cmd "redis-cli ping"
-          --health-interval 5s
-          --health-timeout 5s
-          --health-retries 10
       postgres:
         image: pgvector/pgvector:pg17
         env:
@@ -222,7 +212,6 @@ jobs:
           ORGANIZATIONS_ENABLED: 'true'
           NEXT_PUBLIC_ORGANIZATIONS_ENABLED: 'true'
           INTERNAL_API_SECRET: scim-http-ci-local-secret-at-least-32-characters
-          REDIS_URL: redis://127.0.0.1:6379
           DB_TX_TRIPWIRE: throw
           DISABLE_TELEMETRY: 'true'
           NEXT_TELEMETRY_DISABLED: '1'
@@ -263,6 +252,47 @@ jobs:
           VERSION_COMPARE_E2E_AUTH_SECRET="$BETTER_AUTH_SECRET" \
           VERSION_COMPARE_E2E_REPORT_PATH="$report_dir/version-compare-http-report.json" \
             bun run test:workflow-version-compare:e2e
+
+      # A self-hosted app: hosted billing admits a run only through a Redis usage
+      # reservation, and the SCIM suite above asserts PostgreSQL rate-limit storage,
+      # so workflow execution gets its own app rather than adding Redis to that one.
+      - name: Verify single-block workflow runs over real HTTP
+        working-directory: apps/sim
+        env:
+          NEXT_PUBLIC_APP_URL: http://127.0.0.1:3018
+          BETTER_AUTH_URL: http://127.0.0.1:3018
+          NEXT_PUBLIC_FORCE_HOSTED: 'false'
+          INTERNAL_API_SECRET: stop-after-http-ci-local-secret-at-least-32-characters
+          DB_TX_TRIPWIRE: throw
+          DISABLE_TELEMETRY: 'true'
+          NEXT_TELEMETRY_DISABLED: '1'
+          NEXT_PUBLIC_CHAT_DISABLED: 'true'
+          READY_TIMEOUT_SECONDS: 300
+        run: |
+          report_dir="$RUNNER_TEMP/e2e"
+          server_log="$report_dir/stop-after-next.log"
+          mkdir -p "$report_dir"
+          node ../../node_modules/next/dist/bin/next dev --hostname 127.0.0.1 --port 3018 > "$server_log" 2>&1 &
+          server_pid=$!
+          finish() {
+            kill "$server_pid" 2>/dev/null || true
+            wait "$server_pid" 2>/dev/null || true
+            awk '/^ (GET|POST|PUT|PATCH|DELETE|HEAD) \/api\// { print }' "$server_log" > "$report_dir/stop-after-http-status.log"
+          }
+          trap finish EXIT
+          fail_startup() {
+            echo "::error::$1"
+            tail -n 200 "$server_log"
+            exit 1
+          }
+          started=$SECONDS
+          until curl --fail --silent --max-time 10 http://127.0.0.1:3018/api/health > /dev/null; do
+            kill -0 "$server_pid" 2>/dev/null || fail_startup 'Local workflow app exited during startup.'
+            [ $((SECONDS - started)) -lt "$READY_TIMEOUT_SECONDS" ] ||
+              fail_startup "Local workflow app did not become ready within $READY_TIMEOUT_SECONDS seconds."
+            sleep 2
+          done
+          echo "Local workflow app ready after $((SECONDS - started))s"
           STOP_AFTER_E2E_BASE_URL="$NEXT_PUBLIC_APP_URL" \
           STOP_AFTER_E2E_DATABASE_URL="$DATABASE_URL" \
           STOP_AFTER_E2E_REPORT_PATH="$report_dir/stop-after-http-report.json" \

apps/sim/lib/workflows/application/execute-manual-workflow.test.ts

@@ -306,7 +306,7 @@ describe('manual workflow execution application operations', () => {
         principal,
         input: { ...baseInput, useMockPayload: false, stopAfterBlockId: 'missing' },
       })
-    ).rejects.toMatchObject({ code: 'validation' })
+    ).rejects.toMatchObject({ code: 'validation', message: expect.stringContaining('not a block') })
     await expect(
       executeManualWorkflowFromBlockOperation.execute({
         principal,
@@ -317,7 +317,7 @@ describe('manual workflow execution application operations', () => {
           stopAfterBlockId: 'missing',
         },
       })
-    ).rejects.toMatchObject({ code: 'validation' })
+    ).rejects.toMatchObject({ code: 'validation', message: expect.stringContaining('not a block') })
     expect(mocks.loadSourceState).not.toHaveBeenCalled()
     expect(mocks.executeService).not.toHaveBeenCalled()
   })
@@ -337,38 +337,54 @@ describe('manual workflow execution application operations', () => {
         principal,
         input: { ...baseInput, useMockPayload: false, stopAfterBlockId: 'agent-1' },
       })
-    ).rejects.toMatchObject({ code: 'validation' })
+    ).rejects.toMatchObject({ code: 'validation', message: expect.stringContaining('inside loop') })
     expect(mocks.executeService).not.toHaveBeenCalled()
   })
 
-  it('stops the trigger and block entries after the requested block', async () => {
-    mocks.loadSourceState.mockResolvedValueOnce({ blockStates: {}, executedBlocks: [] })
+  it('rejects a stop block the run cannot reach from its entry, which would run everything', async () => {
+    mockLoadManualState.mockResolvedValue({
+      blocks: { 'trigger-1': {}, 'agent-1': {}, 'agent-2': {}, unconnected: {} },
+      edges: [
+        { source: 'trigger-1', target: 'agent-1' },
+        { source: 'agent-1', target: 'agent-2' },
+      ],
+    })
 
-    await executeManualWorkflowOperation.execute({
-      principal,
-      input: { ...baseInput, useMockPayload: false, stopAfterBlockId: 'agent-1' },
+    await expect(
+      executeManualWorkflowFromBlockOperation.execute({
+        principal,
+        input: {
+          ...baseInput,
+          blockId: 'agent-2',
+          sourceRunId: 'source-run-1',
+          stopAfterBlockId: 'agent-1',
+        },
+      })
+    ).rejects.toMatchObject({
+      code: 'validation',
+      message: expect.stringContaining('not reachable'),
     })
-    await executeManualWorkflowFromBlockOperation.execute({
-      principal,
-      input: {
-        ...baseInput,
-        blockId: 'agent-1',
-        sourceRunId: 'source-run-1',
-        stopAfterBlockId: 'agent-1',
-      },
+    await expect(
+      executeManualWorkflowOperation.execute({
+        principal,
+        input: { ...baseInput, useMockPayload: false, stopAfterBlockId: 'unconnected' },
+      })
+    ).rejects.toMatchObject({
+      code: 'validation',
+      message: expect.stringContaining('not reachable'),
     })
+    expect(mocks.loadSourceState).not.toHaveBeenCalled()
+    expect(mocks.executeService).not.toHaveBeenCalled()
+  })
 
-    expect(mocks.executeService).toHaveBeenNthCalledWith(
-      1,
-      expect.objectContaining({ triggerBlockId: 'trigger-1', stopAfterBlockId: 'agent-1' })
-    )
-    expect(mocks.executeService).toHaveBeenNthCalledWith(
-      2,
-      expect.objectContaining({
-        runFromBlock: expect.objectContaining({ startBlockId: 'agent-1' }),
-        stopAfterBlockId: 'agent-1',
+  it('rejects a stop block named by an inherited object key', async () => {
+    await expect(
+      executeManualWorkflowOperation.execute({
+        principal,
+        input: { ...baseInput, useMockPayload: false, stopAfterBlockId: 'toString' },
       })
-    )
+    ).rejects.toMatchObject({ code: 'validation', message: expect.stringContaining('not a block') })
+    expect(mocks.executeService).not.toHaveBeenCalled()
   })
 
   it('rejects a source run without persisted state for this workflow', async () => {

apps/sim/lib/workflows/application/execute-manual-workflow.ts

@@ -50,15 +50,40 @@ async function loadManualState(workflowId: string) {
 
 type ManualWorkflowState = Awaited<ReturnType<typeof loadManualState>>
 
+function reachableFrom(state: ManualWorkflowState, entryBlockId: string): Set<string> {
+  const targetsBySource = new Map<string, string[]>()
+  for (const edge of state.edges) {
+    const targets = targetsBySource.get(edge.source) ?? []
+    targets.push(edge.target)
+    targetsBySource.set(edge.source, targets)
+  }
+  const reached = new Set([entryBlockId])
+  const queue = [entryBlockId]
+  for (let next = queue.pop(); next !== undefined; next = queue.pop()) {
+    for (const target of targetsBySource.get(next) ?? []) {
+      if (reached.has(target)) continue
+      reached.add(target)
+      queue.push(target)
+    }
+  }
+  return reached
+}
+
 /**
  * The engine stops only when it completes a node whose id equals the target, so
- * an unknown id would silently run the whole workflow, and a block inside a loop
- * or parallel would stop after its first iteration or never. Both are refused,
- * matching the editor, which offers "Run until block" only outside subflows.
+ * a target the run cannot reach would silently run everything after the entry:
+ * an unknown id, a block upstream of the entry, or a block inside a loop or
+ * parallel (which would stop after its first iteration or never). All are
+ * refused, matching the editor, which offers "Run until block" only outside
+ * subflows.
  */
-function assertStopAfterBlock(state: ManualWorkflowState, blockId: string | undefined): void {
+function assertStopAfterBlock(
+  state: ManualWorkflowState,
+  blockId: string | undefined,
+  entryBlockId: string
+): void {
   if (blockId === undefined) return
-  const block = state.blocks[blockId]
+  const block = Object.hasOwn(state.blocks, blockId) ? state.blocks[blockId] : undefined
   if (!block) {
     throw new OrchestrationError(
       'validation',
@@ -71,6 +96,12 @@ function assertStopAfterBlock(state: ManualWorkflowState, blockId: string | unde
       `run.stopAfterBlockId "${blockId}" is inside loop or parallel "${block.data.parentId}"; stop after that container instead.`
     )
   }
+  if (!reachableFrom(state, entryBlockId).has(blockId)) {
+    throw new OrchestrationError(
+      'validation',
+      `run.stopAfterBlockId "${blockId}" is not reachable from entry block "${entryBlockId}"; stop after the entry block or one downstream of it.`
+    )
+  }
 }
 
 function listTriggers(options: ReturnType<typeof resolveTriggerRunOptions>): string {
@@ -117,7 +148,6 @@ export const executeManualWorkflowOperation = defineAuthorizedWorkflowUseCase({
       )
     }
     const state = await loadManualState(context.workflowId)
-    assertStopAfterBlock(state, input.stopAfterBlockId)
     const options = resolveTriggerRunOptions(mergeSubblockStateWithValues(state.blocks))
     if (options.length === 0) {
       throw new OrchestrationError(
@@ -143,6 +173,7 @@ export const executeManualWorkflowOperation = defineAuthorizedWorkflowUseCase({
       )
     }
 
+    assertStopAfterBlock(state, input.stopAfterBlockId, selected.triggerBlockId)
     const executionInput = input.useMockPayload ? selected.mockPayload : input.input
     const validation = validateTriggerInput(selected, executionInput)
     if (!validation.ok) {
@@ -169,7 +200,7 @@ export const executeManualWorkflowFromBlockOperation = defineAuthorizedWorkflowU
         `run.entry.blockId "${input.blockId}" is not a block in the current saved workflow.`
       )
     }
-    assertStopAfterBlock(state, input.stopAfterBlockId)
+    assertStopAfterBlock(state, input.stopAfterBlockId, input.blockId)
 
     const sourceSnapshot = await getExecutionStateForWorkflow(input.sourceRunId, context.workflowId)
     if (!sourceSnapshot) {

apps/sim/lib/workflows/executor/execution-core.ts

@@ -907,6 +907,9 @@ async function executeWorkflowCoreImpl(
         resolvedStopAfterBlockId = buildLoopSentinelEndId(stopAfterBlockId)
       } else if (serializedWorkflow.parallels?.[stopAfterBlockId]) {
         resolvedStopAfterBlockId = buildParallelSentinelEndId(stopAfterBlockId)
+      } else if (!serializedWorkflow.blocks.some((block) => block.id === stopAfterBlockId)) {
+        // The engine stops on an exact node id, so an absent target would run everything.
+        throw new Error(`Stop block ${stopAfterBlockId} is not in the workflow being executed`)
       }
     }
 

apps/sim/scripts/test-workflow-stop-after-e2e.ts

@@ -9,6 +9,7 @@ import { assertDisposableTestDatabaseUrl } from '@sim/db/testing/test-infrastruc
 import { createLogger } from '@sim/logger'
 import { sha256Hex } from '@sim/security/hash'
 import { getErrorMessage } from '@sim/utils/errors'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { isRecordLike } from '@sim/utils/object'
 import { truncate } from '@sim/utils/string'
@@ -32,7 +33,8 @@ import { readResponseTextWithLimit } from '@/lib/core/utils/stream-limits'
 const logger = createLogger('WorkflowStopAfterE2E')
 const execFileAsync = promisify(execFile)
 const MAX_RESPONSE_BYTES = 2 * 1024 * 1024
-const REQUEST_TIMEOUT_MS = 120_000
+/** The first execute request cold-compiles the route under `next dev`. */
+const REQUEST_TIMEOUT_MS = 300_000
 const SLOW_SECONDS = 4
 const SLOW_MS = SLOW_SECONDS * 1000
 const startedAt = new Date().toISOString()
@@ -319,6 +321,17 @@ try {
       { run: { source: 'manual', stopAfterBlockId: '' } },
       'BAD_REQUEST'
     )
+    await expectBadRequest(
+      pipeline.workflowId,
+      {
+        run: {
+          source: 'manual',
+          entry: { type: 'block', blockId: pipeline.check, sourceRunId },
+          stopAfterBlockId: pipeline.slow,
+        },
+      },
+      'BAD_REQUEST'
+    )
     await expectBadRequest(
       pipeline.workflowId,
       { run: { source: 'manual', stopAfterBlockId: pipeline.check }, async: true },
@@ -360,8 +373,28 @@ try {
 } finally {
   try {
     await check('remove disposable fixtures', async () => {
-      await sql`delete from workspace where id = ${workspaceId}`
-      await sql`delete from "user" where id = ${ownerId}`
+      // A response returns before its run finishes persisting logs and large-value
+      // references; a cascade delete racing those writes can be chosen as a deadlock victim.
+      const workflowIds = [pipeline.workflowId, otherPipeline.workflowId]
+      for (let attempt = 0; attempt < 120; attempt++) {
+        const [{ open }] =
+          await sql`select count(*)::int as open from workflow_execution_logs where workflow_id in ${sql(workflowIds)} and ended_at is null`
+        if (open === 0) break
+        await sleep(500)
+      }
+      for (let attempt = 1; ; attempt++) {
+        try {
+          await sql.begin(async (tx) => {
+            await tx`delete from workspace where id = ${workspaceId}`
+            await tx`delete from "user" where id = ${ownerId}`
+          })
+          break
+        } catch (error) {
+          const deadlocked = isRecordLike(error) && error.code === '40P01'
+          if (!deadlocked || attempt === 5) throw error
+          await sleep(1000)
+        }
+      }
       if (directory) await rm(directory, { recursive: true, force: true })
     })
   } catch (error) {

packages/sim-cli/src/commands/protocol/workflow-run-follow.test.ts

@@ -243,6 +243,13 @@ describe('sim workflows run --follow', () => {
     })
   })
 
+  it('refuses an empty --stop-after rather than running the whole draft', async () => {
+    await expect(run(WORKFLOW_ID, '--stop-after', '')).rejects.toThrow(
+      '--stop-after requires a block ID'
+    )
+    expect(requestRaw).not.toHaveBeenCalled()
+  })
+
   it('refuses --stop-after with --async before sending anything', async () => {
     await expect(run(WORKFLOW_ID, '--stop-after', 'agent-1', '--async')).rejects.toThrow(
       'Manual execution does not support --async'

packages/sim-cli/src/commands/protocol/workflow-run-follow.ts

@@ -79,8 +79,11 @@ export function resolveWorkflowRunSelection(
   if (useMockPayload && flags.input !== undefined) {
     throw new SimApiError('--mock-payload cannot be combined with --input', 0)
   }
+  if (stopAfter !== undefined && stopAfter.trim() === '') {
+    throw new SimApiError('--stop-after requires a block ID', 0)
+  }
 
-  const stop = stopAfter ? { stopAfterBlockId: stopAfter } : {}
+  const stop = stopAfter !== undefined ? { stopAfterBlockId: stopAfter } : {}
   if (fromBlock && sourceRun) {
     return {
       source: 'manual',