Repository: PostHog/posthog
Pull request: feat(messaging): Hogflow logical actions, metrics
## Problem

Closes https://github.com/PostHog/posthog/issues/32442
Closes https://github.com/PostHog/posthog/issues/32439


## Changes

tl;dr - adds all logical actions (time and condition based)

- [x] Adds a common delay helper as it will be used across multiple cases
- [x] Makes the duration more advanced with support for the common style we use elsewhere e.g. `1.5h`
- [x] Fixes the delay - we don't need each action to do the default of finding the next one
- [x] Adds 'wait_until_condition', 'delay' 'wait_until_time_window', 'exit' support
- [x] Adds filtering support per action
- [x] Introduces a builder pattern for building the hogflow fixture (we should generally consider this approach all over the code)
- [x] Adds edges and actions to make it easier to build the workflow structures for testing
- [x] Updates all actions to correctly use the edge configurations for branching and continuing



## Did you write or update any docs for this change?

<!-- Engineers are responsible for doing the first pass at documenting their features and/or code.  -->

- [ ] I've [added or updated the docs](https://posthog.com/handbook/engineering/writing-docs)
- [ ] I've reached out for help from the docs team
- [ ] No docs needed for this change

## How did you test this code?

<!-- Briefly describe the steps you took. -->
<!-- Include automated tests if possible, otherwise describe the manual testing routine. -->


Problem statement:
Hogflow metrics + logs
HogFunctions have metrics and logs - we need to extend this to work for HogFlows.

General per-flow invocation metrics and logs will work just fine. Making them be "action" specific is a little trickier... We would either need to add another dimension to the tables for filtering, or do it some hackier way such as combining "workflow_id:action_id" into the ID field


## Idea

* We could double track metrics - one for the specific invocation

Hogflow step executor
- [x] Generic Hogflow Node executor + implementations of logical and time based nodes

Hints from issue discussion:

Commit message: Fixes

Patch to review:
diff --git a/plugin-server/src/cdp/_tests/fixtures-hogflows.ts b/plugin-server/src/cdp/_tests/fixtures-hogflows.ts
index 3859b7dd91664..3521c4dd4be56 100644
--- a/plugin-server/src/cdp/_tests/fixtures-hogflows.ts
+++ b/plugin-server/src/cdp/_tests/fixtures-hogflows.ts
@@ -1,12 +1,13 @@
 import { randomUUID } from 'crypto'
 
-import { HogFlow } from '~/schema/hogflow'
+import { HogFlow, HogFlowAction } from '~/schema/hogflow'
 import { insertRow } from '~/tests/helpers/sql'
 
 import { Team } from '../../types'
 import { PostgresRouter } from '../../utils/db/postgres'
 import { UUIDT } from '../../utils/utils'
 import { CyclotronJobInvocationHogFlow, HogFlowInvocationContext } from '../types'
+import { createHogExecutionGlobals } from './fixtures'
 
 export const createHogFlow = (hogFlow: Partial<HogFlow>) => {
     const item: HogFlow = {
@@ -28,6 +29,24 @@ export const createHogFlow = (hogFlow: Partial<HogFlow>) => {
     return item
 }
 
+export const createHogFlowAction = <T extends HogFlowAction['type']>(
+    overrides: Pick<Extract<HogFlowAction, { type: T }>, 'type' | 'config'> &
+        Partial<Omit<Extract<HogFlowAction, { type: T }>, 'type' | 'config'>>
+): Extract<HogFlowAction, { type: T }> => {
+    const action = {
+        id: randomUUID(),
+        name: 'Action',
+        description: 'Test action',
+        on_error: 'continue',
+        created_at: new Date().getTime(),
+        updated_at: new Date().getTime(),
+        ...overrides,
+    } as unknown as Extract<HogFlowAction, { type: T }>
+    // NOTE(bw): The type cast here is nasty but the getting the type inference correct is beyond me right now!
+
+    return action
+}
+
 export const insertHogFlow = async (
     postgres: PostgresRouter,
     team_id: Team['id'],
@@ -52,6 +71,10 @@ export const createHogFlowInvocationContext = (
     data: Partial<HogFlowInvocationContext> = {}
 ): HogFlowInvocationContext => {
     return {
+        event: {
+            ...createHogExecutionGlobals().event,
+            ...data.event,
+        },
         ...data,
     }
 }
diff --git a/plugin-server/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts b/plugin-server/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts
index 3ac29d3380efb..0da4273fb4017 100644
--- a/plugin-server/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts
+++ b/plugin-server/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts
@@ -12,7 +12,7 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker {
 
     public async processInvocations(invocations: CyclotronJobInvocation[]): Promise<CyclotronJobInvocationResult[]> {
         const loadedInvocations = await this.loadHogFlows(invocations)
-        return await this.runManyWithHeartbeat(loadedInvocations, (item) => this.hogFlowExecutor.execute(item))
+        return await Promise.all(loadedInvocations.map((item) => this.hogFlowExecutor.execute(item)))
     }
 
     protected async loadHogFlows(invocations: CyclotronJobInvocation[]): Promise<CyclotronJobInvocationHogFlow[]> {
diff --git a/plugin-server/src/cdp/consumers/cdp-cyclotron-worker.test.ts b/plugin-server/src/cdp/consumers/cdp-cyclotron-worker.test.ts
index 4139cec24d20c..0d4ea4ee38b6e 100644
--- a/plugin-server/src/cdp/consumers/cdp-cyclotron-worker.test.ts
+++ b/plugin-server/src/cdp/consumers/cdp-cyclotron-worker.test.ts
@@ -185,7 +185,6 @@ describe('CdpCyclotronWorker', () => {
             invocation.functionId = new UUIDT().toString()
             const results = await processor.processInvocations([invocation])
             expect(results).toEqual([])
-
             expect(dequeueInvocationsSpy).toHaveBeenCalledWith([invocation])
         })
     })
diff --git a/plugin-server/src/cdp/consumers/cdp-events-consumer.test.ts b/plugin-server/src/cdp/consumers/cdp-events-consumer.test.ts
index 76a6e23c6cba5..4cce09f9162e8 100644
--- a/plugin-server/src/cdp/consumers/cdp-events-consumer.test.ts
+++ b/plugin-server/src/cdp/consumers/cdp-events-consumer.test.ts
@@ -17,6 +17,8 @@ import {
 import { CdpEventsConsumer } from './cdp-events.consumer'
 import { CdpInternalEventsConsumer } from './cdp-internal-event.consumer'
 import { CyclotronJobQueue } from '../services/job-queue/job-queue'
+import { insertHogFlow as _insertHogFlow } from '../_tests/fixtures-hogflows'
+import { HogFlow } from '~/schema/hogflow'
 
 jest.setTimeout(1000)
 
@@ -298,3 +300,117 @@ describe.each([
         })
     })
 })
+
+describe('hog flow processing', () => {
+    let processor: CdpEventsConsumer | CdpInternalEventsConsumer
+    let hub: Hub
+    let team: Team
+
+    const insertHogFlow = async (hogFlow: Partial<HogFlow>) => {
+        const teamId = hogFlow.team_id ?? team.id
+        const item = await _insertHogFlow(hub.postgres, teamId, {
+            ...hogFlow,
+        })
+        // Trigger the reload that django would do
+        processor['hogFunctionManager']['onHogFunctionsReloaded'](teamId, [item.id])
+        return item
+    }
+
+    beforeEach(async () => {
+        await resetTestDatabase()
+        hub = await createHub()
+        team = await getFirstTeam(hub)
+        processor = new CdpEventsConsumer(hub)
+
+        // NOTE: We don't want to actually connect to Kafka for these tests as it is slow and we are testing the core logic only
+        processor['kafkaConsumer'] = {
+            connect: jest.fn(),
+            disconnect: jest.fn(),
+            isHealthy: jest.fn(),
+        } as any
+
+        processor['cyclotronJobQueue'] = {
+            queueInvocations: jest.fn(),
+            startAsProducer: jest.fn(() => Promise.resolve()),
+            stop: jest.fn(),
+        } as unknown as jest.Mocked<CyclotronJobQueue>
+
+        await processor.start()
+    })
+
+    afterEach(async () => {
+        jest.setTimeout(10000)
+        await processor.stop()
+        await closeHub(hub)
+    })
+
+    afterAll(() => {
+        jest.useRealTimers()
+    })
+
+    describe('createHogFlowInvocations', () => {
+        let globals: HogFunctionInvocationGlobals
+
+        beforeEach(() => {
+            globals = createHogExecutionGlobals({
+                project: {
+                    id: team.id,
+                } as any,
+                event: {
+                    uuid: 'b3a1fe86-b10c-43cc-acaf-d208977608d0',
+                    event: '$pageview',
+                    properties: {
+                        $current_url: 'https://posthog.com',
+                        $lib_version: '1.0.0',
+                    },
+                } as any,
+            })
+        })
+
+        it('should not create hog flow invocations with no filters', async () => {
+            await insertHogFlow({})
+
+            const invocations = await processor['createHogFlowInvocations']([globals])
+            expect(invocations).toHaveLength(0)
+        })
+
+        it('should create hog flow invocations with matching filters', async () => {
+            const hogFlow = await insertHogFlow({
+                trigger: {
+                    type: 'event',
+                    filters: HOG_FILTERS_EXAMPLES.pageview_or_autocapture_filter.filters,
+                },
+            })
+
+            const noInvocations = await processor['createHogFlowInvocations']([
+                {
+                    ...globals,
+                    event: {
+                        ...globals.event,
+                        event: 'not-a-pageview',
+                    },
+                },
+            ])
+
+            expect(noInvocations).toHaveLength(0)
+
+            const invocations = await processor['createHogFlowInvocations']([globals])
+            expect(invocations).toHaveLength(1)
+            expect(invocations[0]).toMatchObject({
+                functionId: hogFlow.id,
+                hogFlow: {
+                    id: hogFlow.id,
+                },
+                id: expect.any(String),
+                queue: 'hogflow',
+                queuePriority: 1,
+                state: {
+                    event: globals.event,
+                    personId: 'uuid',
+                    variables: {},
+                },
+                teamId: 2,
+            })
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflow-executor.service.test.ts b/plugin-server/src/cdp/services/hogflow-executor.service.test.ts
new file mode 100644
index 0000000000000..05f8056730062
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflow-executor.service.test.ts
@@ -0,0 +1,81 @@
+import { HogFlow } from '~/schema/hogflow'
+
+import { Hub } from '../../../src/types'
+import { createHub } from '../../../src/utils/db/hub'
+import { HOG_FILTERS_EXAMPLES } from '../_tests/examples'
+import { createExampleHogFlowInvocation, createHogFlow, createHogFlowAction } from '../_tests/fixtures-hogflows'
+import { HogFlowExecutorService } from './hogflow-executor.service'
+
+describe('Hogflow Executor', () => {
+    jest.setTimeout(1000)
+    let executor: HogFlowExecutorService
+    let hub: Hub
+
+    beforeEach(async () => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2024-06-07T12:00:00.000Z').getTime())
+        hub = await createHub()
+        executor = new HogFlowExecutorService(hub)
+    })
+
+    describe('general event processing', () => {
+        let hogFlow: HogFlow
+
+        beforeEach(() => {
+            hogFlow = createHogFlow({
+                actions: [
+                    createHogFlowAction({
+                        id: '1',
+                        type: 'trigger',
+                        config: {
+                            filters: HOG_FILTERS_EXAMPLES.no_filters.filters,
+                        },
+                    }),
+                ],
+            })
+        })
+
+        it('can execute an invocation', async () => {
+            const invocation = createExampleHogFlowInvocation(hogFlow)
+
+            const result = await executor.execute(invocation)
+            expect(result).toEqual({
+                capturedPostHogEvents: [],
+                invocation: {
+                    state: {
+                        actionStepCount: 0,
+                        currentAction: {
+                            id: '1',
+                            startedAtTimestamp: expect.any(Number),
+                        },
+                        event: {
+                            distinct_id: 'distinct_id',
+                            elements_chain: '',
+                            event: 'test',
+                            properties: {
+                                $lib_version: '1.2.3',
+                            },
+                            timestamp: '2024-06-07T12:00:00.000Z',
+                            url: 'http://localhost:8000/events/1',
+                            uuid: 'uuid',
+                        },
+                    },
+                    id: expect.any(String),
+                    teamId: 1,
+                    hogFlow: invocation.hogFlow,
+                    functionId: invocation.hogFlow.id,
+                    queue: 'hogflow',
+                    queueMetadata: undefined,
+                    queueScheduledAt: undefined,
+                    queueSource: undefined,
+                    queueParameters: undefined,
+                    queuePriority: 0,
+                },
+                error: 'Action type trigger not supported',
+                finished: true,
+                logs: expect.any(Array),
+                metrics: [],
+            })
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflow-executor.service.ts b/plugin-server/src/cdp/services/hogflow-executor.service.ts
index 30164433109f8..a19f8c6440e3e 100644
--- a/plugin-server/src/cdp/services/hogflow-executor.service.ts
+++ b/plugin-server/src/cdp/services/hogflow-executor.service.ts
@@ -1,5 +1,7 @@
+import { DateTime } from 'luxon'
+
 import { HogFlow } from '../../schema/hogflow'
-import { PluginsServerConfig } from '../../types'
+import { Hub } from '../../types'
 import { logger } from '../../utils/logger'
 import { UUIDT } from '../../utils/utils'
 import {
@@ -13,6 +15,9 @@ import {
 import { convertToHogFunctionFilterGlobal } from '../utils'
 import { filterFunctionInstrumented } from '../utils/hog-function-filtering'
 import { createInvocationResult } from '../utils/invocation-utils'
+import { HogFlowActionRunner } from './hogflows/actions'
+
+export const MAX_ACTION_STEPS_HARD_LIMIT = 1000
 
 export function createHogFlowInvocation(
     globals: HogFunctionInvocationGlobals,
@@ -34,7 +39,11 @@ export function createHogFlowInvocation(
 }
 
 export class HogFlowExecutorService {
-    constructor(private config: PluginsServerConfig) {}
+    private hogFlowActionRunner: HogFlowActionRunner
+
+    constructor(private hub: Hub) {
+        this.hogFlowActionRunner = new HogFlowActionRunner(this.hub)
+    }
 
     buildHogFlowInvocations(
         hogFlows: HogFlow[],
@@ -81,21 +90,85 @@ export class HogFlowExecutorService {
         }
     }
 
-    execute(invocation: CyclotronJobInvocationHogFlow): CyclotronJobInvocationResult<CyclotronJobInvocationHogFlow> {
+    async execute(
+        invocation: CyclotronJobInvocationHogFlow
+    ): Promise<CyclotronJobInvocationResult<CyclotronJobInvocationHogFlow>> {
         const loggingContext = {
             invocationId: invocation.id,
             hogFlowId: invocation.hogFlow.id,
             hogFlowName: invocation.hogFlow.name,
         }
 
-        logger.debug('🦔', `[HogExecutor] Executing function`, loggingContext)
+        logger.debug('🦔', `[HogFlowExecutor] Executing hog flow`, loggingContext)
 
-        const result = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {
-            queue: 'hogflow',
-        })
+        const result = createInvocationResult<CyclotronJobInvocationHogFlow>(
+            invocation,
+            {
+                queue: 'hogflow',
+            },
+            {
+                finished: false,
+            }
+        )
+
+        result.invocation.state.actionStepCount = invocation.state.actionStepCount ?? 0
+
+        // TODO: Add early exit for exit conditions being met
+        // Also load the person info for that and cache it on the invocation in a way that won't get serialized
 
-        // TODO: Implement!
+        // NOTE: Todo this right we likely want to enrich the invocation here or earlier with two things:
+        // 1. The person object
+        // 2. the converted filters object as it is likely to be used by many or all runners
+
+        // TODO: Also derive max action step count from the hog flow
+        try {
+            while (!result.finished && result.invocation.state.actionStepCount < MAX_ACTION_STEPS_HARD_LIMIT) {
+                const actionResult = await this.hogFlowActionRunner.runCurrentAction(result.invocation)
+
+                if (!actionResult.finished) {
+                    // If the result isn't finished we _require_ that there is a `scheduledAt` param in order to delay the result
+                    if (!actionResult.scheduledAt) {
+                        throw new Error('Action result is not finished and no scheduledAt param is provided')
+                    }
+
+                    result.finished = false
+                    result.invocation.queueScheduledAt = actionResult.scheduledAt
+                    // TODO: Do we also want to increment some meta context?
+                    // TODO: Add a log here to indicate it is scheduled for later
+                    break
+                }
+
+                if (actionResult.goToActionId) {
+                    result.invocation.state.actionStepCount = (result.invocation.state.actionStepCount ?? 0) + 1
+                    // Update the state to be going to the next action
+                    result.invocation.state.currentAction = {
+                        id: actionResult.goToActionId,
+                        startedAtTimestamp: DateTime.now().toMillis(),
+                    }
+                    result.finished = false // Nothing new here but just to be sure
+                    // TODO: Add a log here to indicate the outcome
+                    break
+                }
+
+                // The action is finished so we move on
+                // TODO: Add a log here to indicate the action lead to the end of the flow
+                result.finished = true
+                break
+            }
+
+            // NOTE: Purposefully wait for the next tick to ensure we don't block up the event loop
+            await new Promise((resolve) => process.nextTick(resolve))
+        } catch (err) {
+            // The final catch - in this case we are always just logging the final outcome
+            result.error = err.message
+            result.finished = true // Explicitly set to true to prevent infinite loops
+            logger.error(
+                '🦔',
+                `[HogFlowExecutor] Error executing hog flow ${invocation.hogFlow.id} - ${invocation.hogFlow.name}. Event: '${invocation.state.event?.url}'`,
+                err
+            )
+        }
 
-        return result
+        return Promise.resolve(result)
     }
 }
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.test.ts b/plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.test.ts
new file mode 100644
index 0000000000000..d3b859a0ed5ba
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.test.ts
@@ -0,0 +1,123 @@
+import { DateTime } from 'luxon'
+
+import { HOG_FILTERS_EXAMPLES } from '~/cdp/_tests/examples'
+import { createExampleHogFlowInvocation, createHogFlowAction } from '~/cdp/_tests/fixtures-hogflows'
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerConditionalBranch } from './action.conditional_branch'
+
+describe('HogFlowActionRunnerCondition', () => {
+    let runner: HogFlowActionRunnerConditionalBranch
+    let invocation: CyclotronJobInvocationHogFlow
+    let action: Extract<HogFlowAction, { type: 'conditional_branch' }>
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2025-01-01T00:00:00.000Z'))
+
+        runner = new HogFlowActionRunnerConditionalBranch()
+        action = createHogFlowAction({
+            type: 'conditional_branch',
+            config: {
+                conditions: [
+                    {
+                        filter: HOG_FILTERS_EXAMPLES.pageview_or_autocapture_filter.filters,
+                        on_match: 'next-action',
+                    },
+                ],
+            },
+        })
+        invocation = createExampleHogFlowInvocation(
+            {
+                actions: [action],
+            },
+            {
+                currentAction: {
+                    id: action.id,
+                    startedAtTimestamp: DateTime.utc().toMillis(),
+                },
+            }
+        )
+    })
+
+    describe('no matching events', () => {
+        it('should return finished if no matches', async () => {
+            invocation.state.event!.event = 'no-match'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+            })
+        })
+
+        describe('wait logic', () => {
+            it('should handle wait duration and schedule next check', async () => {
+                action.config.delay_duration = '2h'
+                const result = await runner.run(invocation, action)
+                expect(result).toEqual({
+                    finished: false,
+                    // Should schedule for 10 minutes from now
+                    scheduledAt: DateTime.utc().plus({ minutes: 10 }),
+                })
+            })
+
+            it('should not schedule for later than the max wait duration', async () => {
+                action.config.delay_duration = '5m'
+                const result = await runner.run(invocation, action)
+                expect(result).toEqual({
+                    finished: false,
+                    // Should schedule for 5 minutes from now
+                    scheduledAt: DateTime.utc().plus({ minutes: 5 }),
+                })
+            })
+
+            it('should throw error if action started at timestamp is invalid', async () => {
+                invocation.state.currentAction = undefined
+                action.config.delay_duration = '300s'
+                await expect(async () => await runner.run(invocation, action)).rejects.toThrow(
+                    "'startedAtTimestamp' is not set or is invalid"
+                )
+            })
+        })
+    })
+
+    describe('matching events', () => {
+        beforeEach(() => {
+            // These values match the pageview_or_autocapture_filter
+            invocation.state.event!.event = '$pageview'
+            invocation.state.event!.properties = {
+                $current_url: 'https://posthog.com',
+            }
+        })
+
+        it('should match condition and go to action', async () => {
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                goToActionId: 'next-action',
+            })
+        })
+
+        it('should ignore conditions that do not match', async () => {
+            action.config.conditions = [
+                {
+                    filter: HOG_FILTERS_EXAMPLES.elements_text_filter.filters, // No match
+                    on_match: 'action-1',
+                },
+                {
+                    filter: HOG_FILTERS_EXAMPLES.elements_text_filter.filters, // No match
+                    on_match: 'action-2',
+                },
+                {
+                    filter: HOG_FILTERS_EXAMPLES.pageview_or_autocapture_filter.filters, // Match
+                    on_match: 'action-3',
+                },
+            ]
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                goToActionId: 'action-3',
+            })
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.ts b/plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.ts
new file mode 100644
index 0000000000000..a94c7a0760b3e
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.ts
@@ -0,0 +1,55 @@
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { calculatedScheduledAt } from './common/delay'
+import { HogFlowActionRunnerResult } from './types'
+
+const DEFAULT_WAIT_DURATION_SECONDS = 10 * 60
+
+export class HogFlowActionRunnerConditionalBranch {
+    run(
+        invocation: CyclotronJobInvocationHogFlow,
+        action: Extract<HogFlowAction, { type: 'conditional_branch' }>
+    ): Promise<HogFlowActionRunnerResult> {
+        const filterGlobals: HogFunctionFilterGlobals = convertToHogFunctionFilterGlobal({
+            event: invocation.state.event, // TODO: Fix typing
+            groups: {},
+        })
+
+        for (const condition of action.config.conditions) {
+            const filterResults = filterFunctionInstrumented({
+                fn: invocation.hogFlow,
+                filters: condition.filter,
+                filterGlobals,
+                eventUuid: invocation.state.event.uuid,
+            })
+
+            if (filterResults.match) {
+                return Promise.resolve({
+                    finished: true,
+                    goToActionId: condition.on_match,
+                })
+            }
+        }
+
+        if (action.config.delay_duration) {
+            // Calculate the scheduledAt based on the delay duration - max we will wait for is 10 minutes which means we check every 10 minutes until the condition is met
+            const scheduledAt = calculatedScheduledAt(
+                action.config.delay_duration,
+                invocation.state.currentAction?.startedAtTimestamp,
+                DEFAULT_WAIT_DURATION_SECONDS
+            )
+
+            return Promise.resolve({
+                finished: !scheduledAt,
+                scheduledAt: scheduledAt ?? undefined,
+            })
+        }
+
+        return Promise.resolve({
+            finished: true,
+        })
+    }
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.delay.test.ts b/plugin-server/src/cdp/services/hogflows/actions/action.delay.test.ts
new file mode 100644
index 0000000000000..cbedfa134a800
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.delay.test.ts
@@ -0,0 +1,68 @@
+import { DateTime } from 'luxon'
+
+import { createExampleHogFlowInvocation, createHogFlowAction } from '~/cdp/_tests/fixtures-hogflows'
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerDelay } from './action.delay'
+
+describe('HogFlowActionRunnerDelay', () => {
+    let runner: HogFlowActionRunnerDelay
+    let invocation: CyclotronJobInvocationHogFlow
+    let action: Extract<HogFlowAction, { type: 'delay' }>
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2025-01-01T00:00:00.000Z'))
+
+        runner = new HogFlowActionRunnerDelay()
+        action = createHogFlowAction({
+            type: 'delay',
+            config: {
+                delay_duration: '10m',
+            },
+        })
+        invocation = createExampleHogFlowInvocation(
+            {
+                actions: [action],
+            },
+            {
+                currentAction: {
+                    id: action.id,
+                    startedAtTimestamp: DateTime.utc().toMillis(),
+                },
+            }
+        )
+    })
+
+    // NOTE: Most tests are covered in the common delay test file
+    describe('delay step logic', () => {
+        it('should handle wait duration and schedule next check', async () => {
+            action.config.delay_duration = '10m'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: false,
+                // Should schedule for 10 minutes from now
+                scheduledAt: DateTime.utc().plus({ minutes: 10 }),
+            })
+        })
+
+        it('should not schedule for later than the max wait duration', async () => {
+            action.config.delay_duration = '5m'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: false,
+                // Should schedule for 5 minutes from now
+                scheduledAt: DateTime.utc().plus({ minutes: 5 }),
+            })
+        })
+
+        it('should throw error if action started at timestamp is invalid', async () => {
+            invocation.state.currentAction = undefined
+            action.config.delay_duration = '300s'
+            await expect(async () => await runner.run(invocation, action)).rejects.toThrow(
+                "'startedAtTimestamp' is not set or is invalid"
+            )
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.delay.ts b/plugin-server/src/cdp/services/hogflows/actions/action.delay.ts
new file mode 100644
index 0000000000000..342c33d881ee3
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.delay.ts
@@ -0,0 +1,22 @@
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { calculatedScheduledAt } from './common/delay'
+import { HogFlowActionRunnerResult } from './types'
+
+export class HogFlowActionRunnerDelay {
+    run(
+        invocation: CyclotronJobInvocationHogFlow,
+        action: Extract<HogFlowAction, { type: 'delay' }>
+    ): Promise<HogFlowActionRunnerResult> {
+        const scheduledAt = calculatedScheduledAt(
+            action.config.delay_duration,
+            invocation.state.currentAction?.startedAtTimestamp
+        )
+
+        return Promise.resolve({
+            finished: !scheduledAt,
+            scheduledAt: scheduledAt ?? undefined,
+        })
+    }
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.test.ts b/plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.test.ts
new file mode 100644
index 0000000000000..24100bdad6764
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.test.ts
@@ -0,0 +1,117 @@
+import { createExampleHogFlowInvocation, createHogFlowAction } from '~/cdp/_tests/fixtures-hogflows'
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerRandomCohortBranch } from './action.random_cohort_branch'
+
+describe('HogFlowActionRunnerRandomCohortBranch', () => {
+    let runner: HogFlowActionRunnerRandomCohortBranch
+    let invocation: CyclotronJobInvocationHogFlow
+    let action: Extract<HogFlowAction, { type: 'random_cohort_branch' }>
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.spyOn(Math, 'random')
+
+        runner = new HogFlowActionRunnerRandomCohortBranch()
+        action = createHogFlowAction({
+            type: 'random_cohort_branch',
+            config: {
+                cohorts: [
+                    { percentage: 30, on_match: 'cohort-a' },
+                    { percentage: 40, on_match: 'cohort-b' },
+                    { percentage: 30, on_match: 'cohort-c' },
+                ],
+            },
+        })
+        invocation = createExampleHogFlowInvocation(
+            {
+                actions: [action],
+            },
+            {
+                currentAction: {
+                    id: action.id,
+                    startedAtTimestamp: Date.now(),
+                },
+            }
+        )
+    })
+
+    it('should select first cohort when random is in first range', async () => {
+        ;(Math.random as jest.Mock).mockReturnValue(0.2) // 20% - in first cohort range
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'cohort-a',
+        })
+    })
+
+    it('should select second cohort when random is in second range', async () => {
+        ;(Math.random as jest.Mock).mockReturnValue(0.4) // 40% - in second cohort range
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'cohort-b',
+        })
+    })
+
+    it('should select third cohort when random is in third range', async () => {
+        ;(Math.random as jest.Mock).mockReturnValue(0.8) // 80% - in third cohort range
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'cohort-c',
+        })
+    })
+
+    it('should handle edge cases at boundaries', async () => {
+        ;(Math.random as jest.Mock).mockReturnValue(0.3) // Exactly at first boundary
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'cohort-a',
+        })
+        ;(Math.random as jest.Mock).mockReturnValue(0.7) // Exactly at second boundary
+        const result2 = await runner.run(invocation, action)
+        expect(result2).toEqual({
+            finished: true,
+            goToActionId: 'cohort-b',
+        })
+    })
+
+    it('should handle single cohort', async () => {
+        action.config.cohorts = [{ percentage: 100, on_match: 'single-cohort' }]
+        ;(Math.random as jest.Mock).mockReturnValue(0.5)
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'single-cohort',
+        })
+    })
+
+    it('should handle uneven percentages', async () => {
+        action.config.cohorts = [
+            { percentage: 25, on_match: 'cohort-a' },
+            { percentage: 75, on_match: 'cohort-b' },
+        ]
+        ;(Math.random as jest.Mock).mockReturnValue(0.5) // 50% - in second cohort range
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'cohort-b',
+        })
+    })
+
+    it('should fallback to last cohort if percentages dont add up to 100', async () => {
+        action.config.cohorts = [
+            { percentage: 30, on_match: 'cohort-a' },
+            { percentage: 30, on_match: 'cohort-b' },
+        ]
+        ;(Math.random as jest.Mock).mockReturnValue(0.9) // 90% - beyond all defined ranges
+        const result = await runner.run(invocation, action)
+        expect(result).toEqual({
+            finished: true,
+            goToActionId: 'cohort-b',
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.ts b/plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.ts
new file mode 100644
index 0000000000000..e2b2bd27f1218
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.ts
@@ -0,0 +1,31 @@
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerResult } from './types'
+
+type Action = Extract<HogFlowAction, { type: 'random_cohort_branch' }>
+
+export class HogFlowActionRunnerRandomCohortBranch {
+    run(invocation: CyclotronJobInvocationHogFlow, action: Action): Promise<HogFlowActionRunnerResult> {
+        const random = Math.random() * 100 // 0-100
+        let cumulativePercentage = 0
+
+        for (const cohort of action.config.cohorts) {
+            cumulativePercentage += cohort.percentage
+            if (random <= cumulativePercentage) {
+                return Promise.resolve({
+                    finished: true,
+                    goToActionId: cohort.on_match,
+                })
+            }
+        }
+
+        // If we somehow get here (shouldn't happen if percentages add up to 100),
+        // go to the last cohort
+        const lastCohort = action.config.cohorts[action.config.cohorts.length - 1]
+        return Promise.resolve({
+            finished: true,
+            goToActionId: lastCohort.on_match,
+        })
+    }
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.test.ts b/plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.test.ts
new file mode 100644
index 0000000000000..fe309b97d29c5
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.test.ts
@@ -0,0 +1,99 @@
+import { DateTime } from 'luxon'
+
+import { HOG_FILTERS_EXAMPLES } from '~/cdp/_tests/examples'
+import { createExampleHogFlowInvocation, createHogFlowAction } from '~/cdp/_tests/fixtures-hogflows'
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerWaitForCondition } from './action.wait_for_condition'
+
+describe('HogFlowActionRunnerWaitForCondition', () => {
+    let runner: HogFlowActionRunnerWaitForCondition
+    let invocation: CyclotronJobInvocationHogFlow
+    let action: Extract<HogFlowAction, { type: 'wait_until_condition' }>
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2025-01-01T00:00:00.000Z'))
+
+        runner = new HogFlowActionRunnerWaitForCondition()
+        action = createHogFlowAction({
+            type: 'wait_until_condition',
+            config: {
+                condition: {
+                    filter: HOG_FILTERS_EXAMPLES.pageview_or_autocapture_filter.filters,
+                    on_match: 'next-action',
+                },
+                max_wait_duration: '10m',
+            },
+        })
+        invocation = createExampleHogFlowInvocation(
+            {
+                actions: [action],
+            },
+            {
+                currentAction: {
+                    id: action.id,
+                    startedAtTimestamp: DateTime.utc().toMillis(),
+                },
+            }
+        )
+    })
+
+    describe('no matching events', () => {
+        it('should handle wait duration and schedule next check', async () => {
+            action.config.max_wait_duration = '2h'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: false,
+                // Should schedule for 10 minutes from now
+                scheduledAt: DateTime.utc().plus({ minutes: 10 }),
+            })
+        })
+
+        it('should not schedule for later than the max wait duration', async () => {
+            action.config.max_wait_duration = '5m'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: false,
+                // Should schedule for 5 minutes from now
+                scheduledAt: DateTime.utc().plus({ minutes: 5 }),
+            })
+        })
+
+        it('should throw error if action started at timestamp is invalid', async () => {
+            invocation.state.currentAction = undefined
+            action.config.max_wait_duration = '300s'
+            await expect(async () => await runner.run(invocation, action)).rejects.toThrow(
+                "'startedAtTimestamp' is not set or is invalid"
+            )
+        })
+    })
+
+    describe('matching events', () => {
+        beforeEach(() => {
+            // These values match the pageview_or_autocapture_filter
+            invocation.state.event!.event = '$pageview'
+            invocation.state.event!.properties = {
+                $current_url: 'https://posthog.com',
+            }
+        })
+
+        it('should match condition and go to action', async () => {
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                goToActionId: 'next-action',
+            })
+        })
+
+        it('should ignore conditions that do not match', async () => {
+            action.config.condition.filter = HOG_FILTERS_EXAMPLES.elements_text_filter.filters // No match
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: false,
+                scheduledAt: DateTime.utc().plus({ minutes: 10 }),
+            })
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.ts b/plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.ts
new file mode 100644
index 0000000000000..7548134bb1d99
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.ts
@@ -0,0 +1,54 @@
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { calculatedScheduledAt } from './common/delay'
+import { HogFlowActionRunnerResult } from './types'
+
+const DEFAULT_WAIT_DURATION_SECONDS = 10 * 60
+
+// NOTE: This is almost identical to the conditional branch action, but we don't need to check for multiple conditions - we could consolidate
+export class HogFlowActionRunnerWaitForCondition {
+    run(
+        invocation: CyclotronJobInvocationHogFlow,
+        action: Extract<HogFlowAction, { type: 'wait_until_condition' }>
+    ): Promise<HogFlowActionRunnerResult> {
+        const filterGlobals: HogFunctionFilterGlobals = convertToHogFunctionFilterGlobal({
+            event: invocation.state.event, // TODO: Fix typing
+            groups: {},
+        })
+
+        const filterResults = filterFunctionInstrumented({
+            fn: invocation.hogFlow,
+            filters: action.config.condition.filter,
+            filterGlobals,
+            eventUuid: invocation.state.event.uuid,
+        })
+
+        if (filterResults.match) {
+            return Promise.resolve({
+                finished: true,
+                goToActionId: action.config.condition.on_match,
+            })
+        }
+
+        if (action.config.max_wait_duration) {
+            // Calculate the scheduledAt based on the delay duration - max we will wait for is 10 minutes which means we check every 10 minutes until the condition is met
+            const scheduledAt = calculatedScheduledAt(
+                action.config.max_wait_duration,
+                invocation.state.currentAction?.startedAtTimestamp,
+                DEFAULT_WAIT_DURATION_SECONDS
+            )
+
+            return Promise.resolve({
+                finished: !scheduledAt,
+                scheduledAt: scheduledAt ?? undefined,
+            })
+        }
+
+        return Promise.resolve({
+            finished: true,
+        })
+    }
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.test.ts b/plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.test.ts
new file mode 100644
index 0000000000000..343cd0338a985
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.test.ts
@@ -0,0 +1,194 @@
+import { DateTime } from 'luxon'
+
+import { createExampleHogFlowInvocation, createHogFlowAction } from '~/cdp/_tests/fixtures-hogflows'
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerWaitUntilTimeWindow } from './action.wait_until_time_window'
+
+describe('HogFlowActionRunnerWaitUntilTimeWindow', () => {
+    let runner: HogFlowActionRunnerWaitUntilTimeWindow
+    let invocation: CyclotronJobInvocationHogFlow
+    let action: Extract<HogFlowAction, { type: 'wait_until_time_window' }>
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2025-01-01T12:00:00.000Z')) // Wednesday at noon UTC
+
+        runner = new HogFlowActionRunnerWaitUntilTimeWindow()
+        action = createHogFlowAction({
+            type: 'wait_until_time_window',
+            config: {
+                timezone: 'UTC',
+                date: 'any',
+                time: ['14:00', '16:00'],
+            },
+        })
+        invocation = createExampleHogFlowInvocation(
+            {
+                actions: [action],
+            },
+            {
+                currentAction: {
+                    id: action.id,
+                    startedAtTimestamp: DateTime.utc().toMillis(),
+                },
+            }
+        )
+    })
+
+    describe('time window scheduling', () => {
+        it('should schedule for today if time window is in the future', async () => {
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 14, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should schedule immediately if current time is within window', async () => {
+            jest.setSystemTime(new Date('2025-01-01T15:00:00.000Z')) // Middle of window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 15, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should schedule for tomorrow if time window has passed', async () => {
+            jest.setSystemTime(new Date('2025-01-01T17:00:00.000Z')) // After time window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 1 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should handle "any" time', async () => {
+            action.config.time = 'any'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 1 }).startOf('day'),
+            })
+        })
+
+        it('should handle time window spanning midnight', async () => {
+            action.config.time = ['23:00', '01:00']
+            jest.setSystemTime(new Date('2025-01-01T22:00:00.000Z')) // Before window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 23, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should handle time window with minutes', async () => {
+            action.config.time = ['14:30', '15:45']
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 14, minute: 30, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should handle time window with minutes when current time is within window', async () => {
+            action.config.time = ['14:30', '15:45']
+            jest.setSystemTime(new Date('2025-01-01T15:00:00.000Z')) // Middle of window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 15, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+    })
+
+    describe('date restrictions', () => {
+        it('should handle weekday restriction', async () => {
+            action.config.date = 'weekday'
+            jest.setSystemTime(new Date('2025-01-04T17:00:00.000Z')) // Saturday
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 2 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }), // Monday
+            })
+        })
+
+        it('should handle weekend restriction', async () => {
+            action.config.date = 'weekend'
+            jest.setSystemTime(new Date('2025-01-01T17:00:00.000Z')) // Wednesday
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 3 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }), // Saturday
+            })
+        })
+
+        it('should handle specific days', async () => {
+            action.config.date = ['monday', 'wednesday', 'friday']
+            jest.setSystemTime(new Date('2025-01-01T17:00:00.000Z')) // Wednesday
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 2 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }), // Friday
+            })
+        })
+
+        it('should handle single specific day', async () => {
+            action.config.date = ['monday']
+            jest.setSystemTime(new Date('2025-01-01T17:00:00.000Z')) // Wednesday
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 5 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }), // Monday
+            })
+        })
+
+        it('should handle consecutive days', async () => {
+            action.config.date = ['monday', 'tuesday', 'wednesday']
+            jest.setSystemTime(new Date('2025-01-01T17:00:00.000Z')) // Wednesday
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 5 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }), // Monday
+            })
+        })
+    })
+
+    describe('timezone handling', () => {
+        it('should respect timezone setting', async () => {
+            action.config.timezone = 'America/New_York'
+            const result = await runner.run(invocation, action)
+            // Compare to UTC for easier understanding
+            expect(DateTime.utc().setZone('America/New_York').toISO()).toMatchInlineSnapshot(
+                `"2025-01-01T07:00:00.000-05:00"`
+            )
+            expect(result.scheduledAt!.toISO()).toMatchInlineSnapshot(`"2025-01-01T14:00:00.000-05:00"`)
+            expect(result.finished).toBe(true)
+        })
+
+        it('should handle timezone with DST', async () => {
+            action.config.timezone = 'America/New_York'
+            jest.setSystemTime(new Date('2025-07-01T12:00:00.000Z')) // Summer time
+            const result = await runner.run(invocation, action)
+            // Compare to UTC for easier understanding
+            expect(DateTime.utc().setZone('America/New_York').toISO()).toMatchInlineSnapshot(
+                `"2025-07-01T08:00:00.000-04:00"`
+            )
+            expect(result.scheduledAt!.toISO()).toMatchInlineSnapshot(`"2025-07-01T14:00:00.000-04:00"`)
+            expect(result.finished).toBe(true)
+        })
+
+        it('should handle timezone with negative offset', async () => {
+            action.config.timezone = 'Asia/Tokyo'
+
+            const result = await runner.run(invocation, action)
+            // Compare to UTC for easier understanding
+            expect(DateTime.utc().setZone('Asia/Tokyo').toISO()).toMatchInlineSnapshot(
+                `"2025-01-01T21:00:00.000+09:00"`
+            )
+            expect(result.scheduledAt!.toISO()).toMatchInlineSnapshot(`"2025-01-02T14:00:00.000+09:00"`)
+            expect(result.finished).toBe(true)
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.ts b/plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.ts
new file mode 100644
index 0000000000000..bdeaa522beb64
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.ts
@@ -0,0 +1,84 @@
+import { DateTime } from 'luxon'
+
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerResult } from './types'
+
+type Action = Extract<HogFlowAction, { type: 'wait_until_time_window' }>
+
+const DAY_NAMES = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'] as const
+
+export class HogFlowActionRunnerWaitUntilTimeWindow {
+    run(invocation: CyclotronJobInvocationHogFlow, action: Action): Promise<HogFlowActionRunnerResult> {
+        const now = DateTime.utc().setZone(action.config.timezone)
+        const nextTime = this.getNextValidTime(now, action.config)
+
+        return Promise.resolve({
+            finished: true,
+            scheduledAt: nextTime,
+        })
+    }
+
+    private getNextValidTime(now: DateTime, config: Action['config']): DateTime {
+        // If time is 'any', just find next valid day
+        if (config.time === 'any') {
+            return this.getNextValidDay(now, config.date)
+        }
+
+        const [startTime, endTime] = config.time
+        const [startHours, startMinutes] = startTime.split(':').map(Number)
+        const [endHours, endMinutes] = endTime.split(':').map(Number)
+
+        // Try today first
+        let nextTime = now.set({ hour: startHours, minute: startMinutes, second: 0, millisecond: 0 })
+        const endTimeToday = now.set({ hour: endHours, minute: endMinutes, second: 0, millisecond: 0 })
+
+        // If we're within the time window today, schedule for end of window
+        if (now >= nextTime && now <= endTimeToday && this.isValidDay(now, config.date)) {
+            return DateTime.now().setZone(config.timezone)
+        }
+
+        // If time has passed or day doesn't match, find next valid day
+        if (nextTime <= now || !this.isValidDay(nextTime, config.date)) {
+            nextTime = this.getNextValidDay(now, config.date).set({
+                hour: startHours,
+                minute: startMinutes,
+                second: 0,
+                millisecond: 0,
+            })
+        }
+
+        return nextTime
+    }
+
+    private isValidDay(date: DateTime, dateConfig: Action['config']['date']): boolean {
+        if (dateConfig === 'any') {
+            return true
+        }
+
+        const day = date.weekday // 1 = Monday, 7 = Sunday
+        const currentDay = DAY_NAMES[day - 1]
+
+        if (dateConfig === 'weekday') {
+            return day >= 1 && day <= 5
+        }
+        if (dateConfig === 'weekend') {
+            return day === 6 || day === 7
+        }
+        if (Array.isArray(dateConfig)) {
+            return dateConfig.includes(currentDay)
+        }
+        return false
+    }
+
+    private getNextValidDay(now: DateTime, dateConfig: Action['config']['date']): DateTime {
+        let nextDay = now.plus({ days: 1 }).startOf('day')
+
+        while (!this.isValidDay(nextDay, dateConfig)) {
+            nextDay = nextDay.plus({ days: 1 })
+        }
+
+        return nextDay
+    }
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/common/delay.test.ts b/plugin-server/src/cdp/services/hogflows/actions/common/delay.test.ts
new file mode 100644
index 0000000000000..6893a634f20dd
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/common/delay.test.ts
@@ -0,0 +1,79 @@
+import { DateTime } from 'luxon'
+
+import { calculatedScheduledAt } from './delay'
+
+describe('calculatedScheduledAt', () => {
+    let startedAtTimestamp: number
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2025-01-01T00:00:00.000Z'))
+        startedAtTimestamp = DateTime.utc().toMillis()
+    })
+
+    describe('delay duration parsing', () => {
+        it.each([
+            ['1d', { days: 1 }],
+            ['2h', { hours: 2 }],
+            ['30m', { minutes: 30 }],
+            ['45s', { seconds: 45 }],
+            ['1.5h', { hours: 1.5 }],
+        ])('should parse duration %s correctly', (duration, expected) => {
+            const result = calculatedScheduledAt(duration, startedAtTimestamp)
+            expect(result).toEqual(DateTime.utc().plus(expected))
+        })
+
+        it('should throw error for invalid duration format', () => {
+            expect(() => calculatedScheduledAt('invalid', startedAtTimestamp)).toThrow('Invalid duration: invalid')
+        })
+
+        it('should throw error for invalid duration unit', () => {
+            expect(() => calculatedScheduledAt('10x', startedAtTimestamp)).toThrow('Invalid duration: 10x')
+        })
+    })
+
+    describe('delay timing', () => {
+        it.each([
+            ['1m', DateTime.fromISO('2025-01-01T00:00:00.000Z').toUTC().plus({ minutes: 1 })],
+            ['2h', DateTime.fromISO('2025-01-01T00:00:00.000Z').toUTC().plus({ hours: 2 })],
+            ['1d', DateTime.fromISO('2025-01-01T00:00:00.000Z').toUTC().plus({ days: 1 })],
+        ])('should schedule for correct time with duration %s', (duration, expectedTime) => {
+            const result = calculatedScheduledAt(duration, startedAtTimestamp)
+            expect(result).toEqual(expectedTime)
+        })
+
+        it('should return null if delay time has already passed', () => {
+            // Set start time to 1 hour ago
+            const pastTimestamp = DateTime.utc().minus({ hours: 1 }).toMillis()
+            const result = calculatedScheduledAt('30m', pastTimestamp)
+            expect(result).toBeNull()
+
+            const result2 = calculatedScheduledAt('61m', pastTimestamp)
+            expect(result2).toEqual(DateTime.utc().plus({ minutes: 1 }))
+        })
+    })
+
+    describe('max delay duration', () => {
+        it('should use max delay duration if provided and smaller than wait time', () => {
+            const result = calculatedScheduledAt('2h', startedAtTimestamp, 300) // 5 minutes max
+            expect(result).toEqual(DateTime.utc().plus({ seconds: 300 }))
+        })
+
+        it('should use wait time if smaller than max delay duration', () => {
+            const result = calculatedScheduledAt('1m', startedAtTimestamp, 300) // 5 minutes max
+            expect(result).toEqual(DateTime.utc().plus({ minutes: 1 }))
+        })
+    })
+
+    describe('error handling', () => {
+        it('should throw error if startedAtTimestamp is undefined', () => {
+            expect(() => calculatedScheduledAt('1h', undefined)).toThrow(
+                "'startedAtTimestamp' is not set or is invalid"
+            )
+        })
+
+        it('should throw error if startedAtTimestamp is invalid', () => {
+            expect(() => calculatedScheduledAt('1h', 0)).toThrow("'startedAtTimestamp' is not set or is invalid")
+        })
+    })
+})
diff --git a/plugin-server/src/cdp/services/hogflows/actions/common/delay.ts b/plugin-server/src/cdp/services/hogflows/actions/common/delay.ts
new file mode 100644
index 0000000000000..fb3a7ab557d24
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/common/delay.ts
@@ -0,0 +1,69 @@
+import { DateTime, DurationLike } from 'luxon'
+
+// Value is expected to be like `10d` or `1.5h` or `10m`
+
+const DURATION_REGEX = /^(\d*\.?\d+)([dhms])$/
+
+/**
+ * Helper for the common case of delaying a hog flow action.
+ * We calculate the delay value and return the scheduleAt based on the time the action started.
+ * If an optional value is given of the max delay duration, we will use that instead of the default.
+ */
+export function calculatedScheduledAt(
+    value: string,
+    startedAtTimestamp?: number,
+    maxDelaySeconds?: number
+): DateTime | null {
+    const actionStartedAt = startedAtTimestamp ? DateTime.fromMillis(startedAtTimestamp).toUTC() : null
+
+    if (!actionStartedAt || !actionStartedAt.isValid) {
+        throw new Error("'startedAtTimestamp' is not set or is invalid")
+    }
+
+    const match = DURATION_REGEX.exec(value)
+
+    if (!match) {
+        throw new Error(`Invalid duration: ${value}`)
+    }
+
+    const [_, amountString, unit] = match
+
+    let duration: DurationLike
+
+    switch (unit) {
+        case 'd':
+            duration = { days: parseFloat(amountString) }
+            break
+        case 'h':
+            duration = { hours: parseFloat(amountString) }
+            break
+        case 'm':
+            duration = { minutes: parseFloat(amountString) }
+            break
+        case 's':
+            duration = { seconds: parseFloat(amountString) }
+            break
+        default:
+            throw new Error(`Invalid duration: ${value}`)
+    }
+
+    const waitUntilTime = actionStartedAt.plus(duration)
+
+    if (DateTime.utc().diff(waitUntilTime).as('seconds') > 0) {
+        // If the wait until time has already passed, we can just return to indicate no delay is needed
+        return null
+    }
+
+    if (!maxDelaySeconds) {
+        return waitUntilTime
+    }
+
+    // If a max delay seconds is provided, we will use that if smaller than the wait until time
+    let scheduledAt = DateTime.utc().plus({ seconds: maxDelaySeconds })
+
+    if (waitUntilTime.diff(scheduledAt).as('seconds') < 0) {
+        scheduledAt = waitUntilTime
+    }
+
+    return scheduledAt
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/index.ts b/plugin-server/src/cdp/services/hogflows/actions/index.ts
new file mode 100644
index 0000000000000..a9da1fb199739
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/index.ts
@@ -0,0 +1,128 @@
+import { DateTime } from 'luxon'
+
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
+import { Hub } from '~/types'
+import { logger } from '~/utils/logger'
+
+import { HogFlowActionRunnerConditionalBranch } from './action.conditional_branch'
+import { HogFlowActionRunnerDelay } from './action.delay'
+import { HogFlowActionRunnerWaitForCondition } from './action.wait_for_condition'
+import { HogFlowActionRunnerWaitUntilTimeWindow } from './action.wait_until_time_window'
+import { HogFlowActionRunnerResult } from './types'
+
+export class HogFlowActionRunner {
+    private hogFlowActionRunnerConditionalBranch: HogFlowActionRunnerConditionalBranch
+    private hogFlowActionRunnerDelay: HogFlowActionRunnerDelay
+    private hogFlowActionRunnerWaitForCondition: HogFlowActionRunnerWaitForCondition
+    private hogFlowActionRunnerWaitUntilTimeWindow: HogFlowActionRunnerWaitUntilTimeWindow
+
+    constructor(private hub: Hub) {
+        this.hogFlowActionRunnerConditionalBranch = new HogFlowActionRunnerConditionalBranch()
+        this.hogFlowActionRunnerDelay = new HogFlowActionRunnerDelay()
+        this.hogFlowActionRunnerWaitForCondition = new HogFlowActionRunnerWaitForCondition()
+        this.hogFlowActionRunnerWaitUntilTimeWindow = new HogFlowActionRunnerWaitUntilTimeWindow()
+    }
+
+    private findNextActionToRun(_invocation: CyclotronJobInvocationHogFlow): string | undefined {
+        // Finds the next action to be run
+
+        // TODO: Implement this!
+
+        return undefined
+    }
+
+    private shouldSkipAction(invocation: CyclotronJobInvocationHogFlow, action: HogFlowAction): boolean {
+        if (!action.filters) {
+            return false
+        }
+
+        // TODO: Make filterGlobals, person and groups something we load lazily onto the main invocation object to be re-used anywhere
+        // this function isn't super cheap to run
+        const filterGlobals: HogFunctionFilterGlobals = convertToHogFunctionFilterGlobal({
+            event: invocation.state.event, // TODO: Fix typing
+            // TODO: Add person and groups!
+            groups: {},
+        })
+
+        const filterResults = filterFunctionInstrumented({
+            fn: invocation.hogFlow,
+            filters: action.filters,
+            filterGlobals,
+            eventUuid: invocation.state.event.uuid,
+        })
+
+        return filterResults.match
+    }
+
+    async runCurrentAction(invocation: CyclotronJobInvocationHogFlow): Promise<HogFlowActionRunnerResult> {
+        if (!invocation.state.currentAction) {
+            const triggerAction = invocation.hogFlow.actions.find((action) => action.type === 'trigger')
+            if (!triggerAction) {
+                throw new Error('No trigger action found')
+            }
+            // Se the current action to the trigger action
+            invocation.state.currentAction = {
+                id: triggerAction.id,
+                startedAtTimestamp: DateTime.now().toMillis(),
+            }
+
+            // TODO: For the trigger action we need to assume that we have already been "started" this way and move on...
+        }
+
+        const currentActionId = invocation.state.currentAction?.id
+        const action = invocation.hogFlow.actions.find((action) => action.id === currentActionId)
+        if (!action) {
+            throw new Error(`Action ${currentActionId} not found`)
+        }
+
+        if (this.shouldSkipAction(invocation, action)) {
+            // Before we do anything check for filter conditions on the user
+            return {
+                finished: true,
+                goToActionId: this.findNextActionToRun(invocation),
+            }
+        }
+
+        logger.debug('🦔', `[HogFlowActionRunner] Running action ${action.type}`, {
+            action,
+            invocation,
+        })
+
+        let result: HogFlowActionRunnerResult
+
+        switch (action.type) {
+            case 'conditional_branch':
+                result = await this.hogFlowActionRunnerConditionalBranch.run(invocation, action)
+                break
+            case 'delay':
+                result = await this.hogFlowActionRunnerDelay.run(invocation, action)
+                break
+            case 'wait_until_condition':
+                result = await this.hogFlowActionRunnerWaitForCondition.run(invocation, action)
+                break
+            case 'wait_until_time_window':
+                result = await this.hogFlowActionRunnerWaitUntilTimeWindow.run(invocation, action)
+                break
+            case 'exit':
+                result = {
+                    finished: true,
+                }
+                // We are truly finished here
+                return result
+            default:
+                throw new Error(`Action type ${action.type} not supported`)
+        }
+
+        // TODO: If the result is finished and no goToActionId is provided, we need to automatically find the next action to run
+
+        if (result.finished && !result.goToActionId) {
+            // TODO: Find the next action to run
+            result.goToActionId = this.findNextActionToRun(invocation)
+        }
+
+        return result
+    }
+}
diff --git a/plugin-server/src/cdp/services/hogflows/actions/types.ts b/plugin-server/src/cdp/services/hogflows/actions/types.ts
new file mode 100644
index 0000000000000..d3ce6c5625838
--- /dev/null
+++ b/plugin-server/src/cdp/services/hogflows/actions/types.ts
@@ -0,0 +1,8 @@
+import { DateTime } from 'luxon'
+
+// Opinionated version of the CyclotronJobInvocationResult limiting what an action can do
+export type HogFlowActionRunnerResult = {
+    finished: boolean
+    scheduledAt?: DateTime
+    goToActionId?: string
+}
diff --git a/plugin-server/src/cdp/types.ts b/plugin-server/src/cdp/types.ts
index 2e7ca9777d943..559543585cb3d 100644
--- a/plugin-server/src/cdp/types.ts
+++ b/plugin-server/src/cdp/types.ts
@@ -290,10 +290,14 @@ export type CyclotronJobInvocationHogFlow = CyclotronJobInvocation & {
 }
 
 export type HogFlowInvocationContext = {
+    event: HogFunctionInvocationGlobals['event']
     personId?: string
-    event?: any // TODO: Type better
     variables?: Record<string, any>
-    currentActionId?: string
+    currentAction?: {
+        id: string
+        startedAtTimestamp: number
+    }
+    actionStepCount?: number
 }
 
 // Mostly copied from frontend types
diff --git a/plugin-server/src/cdp/utils.ts b/plugin-server/src/cdp/utils.ts
index dd1e8754d27d4..8d537d008b233 100644
--- a/plugin-server/src/cdp/utils.ts
+++ b/plugin-server/src/cdp/utils.ts
@@ -174,7 +174,9 @@ function getElementsChainElements(elementsChain: string): string[] {
     return Array.from(elementMatches)
 }
 
-export function convertToHogFunctionFilterGlobal(globals: HogFunctionInvocationGlobals): HogFunctionFilterGlobals {
+export function convertToHogFunctionFilterGlobal(
+    globals: Pick<HogFunctionInvocationGlobals, 'event' | 'person' | 'groups'>
+): HogFunctionFilterGlobals {
     const groups: Record<string, any> = {}
 
     // We need to add default empty groups so that filtering works as it expects it to always exist
diff --git a/plugin-server/src/cdp/utils/hog-function-filtering.ts b/plugin-server/src/cdp/utils/hog-function-filtering.ts
index 9ca2cacd967f8..76913fdd8ea6b 100644
--- a/plugin-server/src/cdp/utils/hog-function-filtering.ts
+++ b/plugin-server/src/cdp/utils/hog-function-filtering.ts
@@ -39,6 +39,7 @@ export function filterFunctionInstrumented(options: {
 }): HogFilterResult {
     const { fn, filters, filterGlobals, enabledTelemetry, eventUuid } = options
     const type = 'type' in fn ? fn.type : 'hogflow'
+    const fnKind = 'type' in fn ? 'HogFunction' : 'HogFlow'
     const start = performance.now()
     const logs: LogEntry[] = []
     const metrics: MinimalAppMetric[] = []
@@ -77,7 +78,7 @@ export function filterFunctionInstrumented(options: {
             })
         }
     } catch (error) {
-        logger.error('🦔', `[HogFunction] Error filtering function`, {
+        logger.error('🦔', `[${fnKind}] Error filtering function`, {
             functionId: fn.id,
             functionName: fn.name,
             teamId: fn.team_id,
@@ -96,7 +97,7 @@ export function filterFunctionInstrumented(options: {
         if (eventUuid) {
             logs.push({
                 team_id: fn.team_id,
-                log_source: 'hog_function',
+                log_source: fnKind === 'HogFunction' ? 'hog_function' : 'hog_flow',
                 log_source_id: fn.id,
                 instance_id: new UUIDT().toString(),
                 timestamp: DateTime.now(),
@@ -114,7 +115,7 @@ export function filterFunctionInstrumented(options: {
         hogFunctionFilterDuration.observe({ type }, duration)
 
         if (duration > DEFAULT_TIMEOUT_MS) {
-            logger.error('🦔', `[HogFunction] Filter took longer than expected`, {
+            logger.error('🦔', `[${fnKind}] Filter took longer than expected`, {
                 functionId: fn.id,
                 functionName: fn.name,
                 teamId: fn.team_id,
diff --git a/plugin-server/src/schema/hogflow.ts b/plugin-server/src/schema/hogflow.ts
index 00dcedbe9821d..2d44027e52169 100644
--- a/plugin-server/src/schema/hogflow.ts
+++ b/plugin-server/src/schema/hogflow.ts
@@ -1,5 +1,113 @@
 import { z } from 'zod'
 
+const _commonActionFields = {
+    id: z.string(),
+    name: z.string(),
+    description: z.string(),
+    on_error: z.enum(['continue', 'abort', 'complete', 'branch']).optional(),
+    created_at: z.number(),
+    updated_at: z.number(),
+    filters: z.any(), // TODO: Correct to the right type
+}
+
+const HogFlowActionSchema = z.discriminatedUnion('type', [
+    // Trigger
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('trigger'),
+        config: z.object({
+            filters: z.any(),
+        }),
+    }),
+    // Branching
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('conditional_branch'),
+        config: z.object({
+            conditions: z.array(
+                z.object({
+                    filter: z.any(), // type this stronger
+                    on_match: z.string(), // TODO: Can we type this more directly to an edge?
+                })
+            ),
+            delay_duration: z.string().optional(),
+        }),
+    }),
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('random_cohort_branch'),
+        config: z.object({
+            cohorts: z.array(
+                z.object({
+                    percentage: z.number(),
+                    on_match: z.string(), // TODO: Can we type this more directly to an edge?
+                })
+            ),
+        }),
+    }),
+
+    // Time based
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('delay'),
+        config: z.object({
+            delay_duration: z.string(),
+        }),
+    }),
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('wait_until_condition'),
+        config: z.object({
+            condition: z.object({
+                filter: z.any(), // type this stronger
+                on_match: z.string(), // TODO: Can we type this more directly to an edge?
+            }),
+            max_wait_duration: z.string(),
+        }),
+    }),
+
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('wait_until_time_window'),
+        config: z.object({
+            timezone: z.string(),
+            // Date can be special values "weekday", "weekend" or a list of days of the week e.g. 'monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'
+            date: z.union([
+                z.literal('any'),
+                z.literal('weekday'),
+                z.literal('weekend'),
+                z.array(z.enum(['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'])),
+            ]),
+            // time can be "any", or a time range [start, end]
+            time: z.union([
+                z.literal('any'),
+                z.array(z.string()), // e.g. ['10:00', '11:00']
+            ]),
+        }),
+    }),
+    // z.object({
+    //     ..._commonActionFields,
+    //     type: z.literal('message'),
+    //     config: z.object({
+    //         message: z.string(),
+    //         channel: z.string(),
+    //     }),
+    // }),
+    // z.object({
+    //     ..._commonActionFields,
+    //     type: z.literal('hog_function'),
+    //     function_id: z.string(),
+    //     config: z.object({
+    //         args: z.record(z.any()),
+    //     }),
+    // }),
+    z.object({
+        ..._commonActionFields,
+        type: z.literal('exit'),
+        config: z.object({}),
+    }),
+])
+
 export const HogFlowSchema = z.object({
     id: z.string(),
     team_id: z.number(),
@@ -37,26 +145,7 @@ export const HogFlowSchema = z.object({
             index: z.number(),
         })
     ),
-    actions: z.array(
-        z.object({
-            id: z.string(),
-            name: z.string(),
-            description: z.string(),
-            type: z.enum([
-                'trigger',
-                'conditional_branch',
-                'delay',
-                'wait_for_condition',
-                'message',
-                'hog_function',
-                'exit',
-            ]),
-            config: z.any(),
-            on_error: z.enum(['continue', 'abort', 'complete', 'branch']).optional(),
-            created_at: z.number(),
-            updated_at: z.number(),
-        })
-    ),
+    actions: z.array(HogFlowActionSchema),
     abort_action: z.string().optional(),
 })
 


Code review comments:
diff_hunk: @@ -1,19 +1,63 @@
 import { DateTime } from 'luxon'
 
-import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
 import { Hub } from '~/types'
+import { logger } from '~/utils/logger'
 
-import { HogFlowActionRunnerCondition } from './condition.action'
+import { HogFlowActionRunnerConditionalBranch } from './action.conditional_branch'
+import { HogFlowActionRunnerDelay } from './action.delay'
+import { HogFlowActionRunnerWaitForCondition } from './action.wait_for_condition'
+import { HogFlowActionRunnerWaitUntilTimeWindow } from './action.wait_until_time_window'
 import { HogFlowActionRunnerResult } from './types'
 
 export class HogFlowActionRunner {
-    private hogFlowActionRunnerCondition: HogFlowActionRunnerCondition
+    private hogFlowActionRunnerConditionalBranch: HogFlowActionRunnerConditionalBranch
+    private hogFlowActionRunnerDelay: HogFlowActionRunnerDelay
+    private hogFlowActionRunnerWaitForCondition: HogFlowActionRunnerWaitForCondition
+    private hogFlowActionRunnerWaitUntilTimeWindow: HogFlowActionRunnerWaitUntilTimeWindow
 
     constructor(private hub: Hub) {
-        this.hogFlowActionRunnerCondition = new HogFlowActionRunnerCondition()
+        this.hogFlowActionRunnerConditionalBranch = new HogFlowActionRunnerConditionalBranch()
+        this.hogFlowActionRunnerDelay = new HogFlowActionRunnerDelay()
+        this.hogFlowActionRunnerWaitForCondition = new HogFlowActionRunnerWaitForCondition()
+        this.hogFlowActionRunnerWaitUntilTimeWindow = new HogFlowActionRunnerWaitUntilTimeWindow()
     }
 
-    runCurrentAction(invocation: CyclotronJobInvocationHogFlow): Promise<HogFlowActionRunnerResult> {
+    private findNextActionToRun(_invocation: CyclotronJobInvocationHogFlow): string | undefined {
+        // Finds the next action to be run
+
+        // TODO: Implement this!
+
+        return undefined
+    }
path: plugin-server/src/cdp/services/hogflows/actions/index.ts
text: @user1:
logic: findNextActionToRun is a critical function marked as TODO but returning undefined. This could cause flows to terminate unexpectedly when result.goToActionId is needed.

diff_hunk: @@ -27,46 +30,81 @@ const HogFlowActionSchema = z.discriminatedUnion('type', [
                     on_match: z.string(), // TODO: Can we type this more directly to an edge?
                 })
             ),
-            wait_duration_seconds: z.number().optional(),
+            delay_duration: z.string().optional(),
         }),
     }),
     z.object({
         ..._commonActionFields,
-        type: z.literal('delay'),
+        type: z.literal('random_cohort_branch'),
         config: z.object({
-            delay_seconds: z.number(),
+            cohorts: z.array(
+                z.object({
+                    percentage: z.number(),
+                    on_match: z.string(), // TODO: Can we type this more directly to an edge?
+                })
+            ),
         }),
     }),
+
+    // Time based
     z.object({
         ..._commonActionFields,
-        type: z.literal('wait_for_condition'),
+        type: z.literal('delay'),
         config: z.object({
-            condition: z.any(),
-            timeout_seconds: z.number(),
+            delay_duration: z.string(),
         }),
     }),
     z.object({
         ..._commonActionFields,
-        type: z.literal('message'),
+        type: z.literal('wait_until_condition'),
         config: z.object({
-            message: z.string(),
-            channel: z.string(),
+            condition: z.object({
+                filter: z.any(), // type this stronger
+                on_match: z.string(), // TODO: Can we type this more directly to an edge?
+            }),
+            max_wait_duration: z.string(),
         }),
     }),
+
     z.object({
         ..._commonActionFields,
-        type: z.literal('hog_function'),
-        function_id: z.string(),
+        type: z.literal('wait_until_time_window'),
         config: z.object({
-            args: z.record(z.any()),
+            timezone: z.string(),
+            // Date can be special values "weekday", "weekend" or a list of days of the week e.g. 'monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'
+            date: z.union([
+                z.literal('any'),
+                z.literal('weekday'),
+                z.literal('weekend'),
+                z.array(z.enum(['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'])),
+            ]),
+            // time can be "any", or a time range [start, end]
+            time: z.union([
+                z.literal('any'),
+                z.array(z.string()), // e.g. ['10:00', '11:00']
path: plugin-server/src/schema/hogflow.ts
text: @user1:
logic: Time array values need validation to ensure they are valid 24h format strings (HH:mm). Use z.string().regex() with proper time format validation

diff_hunk: @@ -0,0 +1,194 @@
+import { DateTime } from 'luxon'
+
+import { createExampleHogFlowInvocation, createHogFlowAction } from '~/cdp/_tests/fixtures-hogflows'
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerWaitUntilTimeWindow } from './action.wait_until_time_window'
+
+describe('HogFlowActionRunnerWaitUntilTimeWindow', () => {
+    let runner: HogFlowActionRunnerWaitUntilTimeWindow
+    let invocation: CyclotronJobInvocationHogFlow
+    let action: Extract<HogFlowAction, { type: 'wait_until_time_window' }>
+
+    beforeEach(() => {
+        jest.useFakeTimers()
+        jest.setSystemTime(new Date('2025-01-01T12:00:00.000Z')) // Wednesday at noon UTC
+
+        runner = new HogFlowActionRunnerWaitUntilTimeWindow()
+        action = createHogFlowAction({
+            type: 'wait_until_time_window',
+            config: {
+                timezone: 'UTC',
+                date: 'any',
+                time: ['14:00', '16:00'],
+            },
+        })
+        invocation = createExampleHogFlowInvocation(
+            {
+                actions: [action],
+            },
+            {
+                currentAction: {
+                    id: action.id,
+                    startedAtTimestamp: DateTime.utc().toMillis(),
+                },
+            }
+        )
+    })
+
+    describe('time window scheduling', () => {
+        it('should schedule for today if time window is in the future', async () => {
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 14, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should schedule immediately if current time is within window', async () => {
+            jest.setSystemTime(new Date('2025-01-01T15:00:00.000Z')) // Middle of window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 15, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should schedule for tomorrow if time window has passed', async () => {
+            jest.setSystemTime(new Date('2025-01-01T17:00:00.000Z')) // After time window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 1 }).set({ hour: 14, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
+
+        it('should handle "any" time', async () => {
+            action.config.time = 'any'
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().plus({ days: 1 }).startOf('day'),
+            })
+        })
+
+        it('should handle time window spanning midnight', async () => {
+            action.config.time = ['23:00', '01:00']
+            jest.setSystemTime(new Date('2025-01-01T22:00:00.000Z')) // Before window
+            const result = await runner.run(invocation, action)
+            expect(result).toEqual({
+                finished: true,
+                scheduledAt: DateTime.utc().set({ hour: 23, minute: 0, second: 0, millisecond: 0 }),
+            })
+        })
path: plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.test.ts
text: @user1:
logic: Missing edge case: no test verifies behavior when the current time is within a window spanning midnight. Could lead to incorrect scheduling.

diff_hunk: @@ -0,0 +1,31 @@
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerResult } from './types'
+
+type Action = Extract<HogFlowAction, { type: 'random_cohort_branch' }>
+
+export class HogFlowActionRunnerRandomCohortBranch {
+    run(invocation: CyclotronJobInvocationHogFlow, action: Action): Promise<HogFlowActionRunnerResult> {
+        const random = Math.random() * 100 // 0-100
path: plugin-server/src/cdp/services/hogflows/actions/action.random_cohort_branch.ts
text: @user1:
style: Consider using aa8e59ce-b52f-4bd8-af79-3a909ea2e1da: For consistent results across instances, use percentage rollout on person IDs instead of Math.random()

diff_hunk: @@ -27,46 +30,81 @@ const HogFlowActionSchema = z.discriminatedUnion('type', [
                     on_match: z.string(), // TODO: Can we type this more directly to an edge?
                 })
             ),
-            wait_duration_seconds: z.number().optional(),
+            delay_duration: z.string().optional(),
         }),
     }),
     z.object({
         ..._commonActionFields,
-        type: z.literal('delay'),
+        type: z.literal('random_cohort_branch'),
         config: z.object({
-            delay_seconds: z.number(),
+            cohorts: z.array(
+                z.object({
+                    percentage: z.number(),
+                    on_match: z.string(), // TODO: Can we type this more directly to an edge?
+                })
+            ),
         }),
     }),
+
+    // Time based
     z.object({
         ..._commonActionFields,
-        type: z.literal('wait_for_condition'),
+        type: z.literal('delay'),
         config: z.object({
-            condition: z.any(),
-            timeout_seconds: z.number(),
+            delay_duration: z.string(),
         }),
     }),
     z.object({
         ..._commonActionFields,
-        type: z.literal('message'),
+        type: z.literal('wait_until_condition'),
         config: z.object({
-            message: z.string(),
-            channel: z.string(),
+            condition: z.object({
+                filter: z.any(), // type this stronger
+                on_match: z.string(), // TODO: Can we type this more directly to an edge?
+            }),
+            max_wait_duration: z.string(),
         }),
     }),
+
     z.object({
         ..._commonActionFields,
-        type: z.literal('hog_function'),
-        function_id: z.string(),
+        type: z.literal('wait_until_time_window'),
         config: z.object({
-            args: z.record(z.any()),
+            timezone: z.string(),
+            // Date can be special values "weekday", "weekend" or a list of days of the week e.g. 'monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'
+            date: z.union([
+                z.literal('any'),
+                z.literal('weekday'),
+                z.literal('weekend'),
+                z.array(z.enum(['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'])),
+            ]),
+            // time can be "any", or a time range [start, end]
+            time: z.union([
+                z.literal('any'),
+                z.array(z.string()), // e.g. ['10:00', '11:00']
+            ]),
path: plugin-server/src/schema/hogflow.ts
text: @user1:
style: Use z.tuple([z.string(), z.string()]) instead of z.array(z.string()) to enforce exactly 2 time values for the range

diff_hunk: @@ -0,0 +1,55 @@
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { calculatedScheduledAt } from './common/delay'
+import { HogFlowActionRunnerResult } from './types'
+
+const DEFAULT_WAIT_DURATION_SECONDS = 10 * 60
+
+export class HogFlowActionRunnerConditionalBranch {
+    run(
+        invocation: CyclotronJobInvocationHogFlow,
+        action: Extract<HogFlowAction, { type: 'conditional_branch' }>
+    ): Promise<HogFlowActionRunnerResult> {
+        const filterGlobals: HogFunctionFilterGlobals = convertToHogFunctionFilterGlobal({
+            event: invocation.state.event, // TODO: Fix typing
+            groups: {},
+        })
+
+        for (const condition of action.config.conditions) {
+            const filterResults = filterFunctionInstrumented({
+                fn: invocation.hogFlow,
+                filters: condition.filter,
+                filterGlobals,
+                eventUuid: invocation.state.event.uuid,
+            })
+
+            if (filterResults.match) {
+                return Promise.resolve({
+                    finished: true,
+                    goToActionId: condition.on_match,
+                })
+            }
+        }
+
+        if (action.config.delay_duration) {
+            // Calculate the scheduledAt based on the delay duration - max we will wait for is 10 minutes which means we check every 10 minutes until the condition is met
+            const scheduledAt = calculatedScheduledAt(
+                action.config.delay_duration,
+                invocation.state.currentAction?.startedAtTimestamp,
+                DEFAULT_WAIT_DURATION_SECONDS
+            )
+
+            return Promise.resolve({
+                finished: !scheduledAt,
+                scheduledAt: scheduledAt ?? undefined,
path: plugin-server/src/cdp/services/hogflows/actions/action.conditional_branch.ts
text: @user1:
logic: The !scheduledAt logic is inverted - if scheduledAt is null it means no delay is needed (already passed) and should return finished: true

diff_hunk: @@ -0,0 +1,54 @@
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { calculatedScheduledAt } from './common/delay'
+import { HogFlowActionRunnerResult } from './types'
+
+const DEFAULT_WAIT_DURATION_SECONDS = 10 * 60
+
+// NOTE: This is almost identical to the conditional branch action, but we don't need to check for multiple conditions - we could consolidate
+export class HogFlowActionRunnerWaitForCondition {
+    run(
+        invocation: CyclotronJobInvocationHogFlow,
+        action: Extract<HogFlowAction, { type: 'wait_until_condition' }>
+    ): Promise<HogFlowActionRunnerResult> {
+        const filterGlobals: HogFunctionFilterGlobals = convertToHogFunctionFilterGlobal({
+            event: invocation.state.event, // TODO: Fix typing
+            groups: {},
path: plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.ts
text: @user1:
logic: Passing empty groups object ({}) could lead to incorrect filtering if group-based conditions are used. Should pass invocation.state.groups if available.

diff_hunk: @@ -0,0 +1,54 @@
+import { CyclotronJobInvocationHogFlow, HogFunctionFilterGlobals } from '~/cdp/types'
+import { convertToHogFunctionFilterGlobal } from '~/cdp/utils'
+import { filterFunctionInstrumented } from '~/cdp/utils/hog-function-filtering'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { calculatedScheduledAt } from './common/delay'
+import { HogFlowActionRunnerResult } from './types'
+
+const DEFAULT_WAIT_DURATION_SECONDS = 10 * 60
path: plugin-server/src/cdp/services/hogflows/actions/action.wait_for_condition.ts
text: @user1:
style: Consider moving DEFAULT_WAIT_DURATION_SECONDS to a constants file for better maintainability

diff_hunk: @@ -0,0 +1,84 @@
+import { DateTime } from 'luxon'
+
+import { CyclotronJobInvocationHogFlow } from '~/cdp/types'
+import { HogFlowAction } from '~/schema/hogflow'
+
+import { HogFlowActionRunnerResult } from './types'
+
+type Action = Extract<HogFlowAction, { type: 'wait_until_time_window' }>
+
+const DAY_NAMES = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'] as const
+
+export class HogFlowActionRunnerWaitUntilTimeWindow {
+    run(invocation: CyclotronJobInvocationHogFlow, action: Action): Promise<HogFlowActionRunnerResult> {
+        const now = DateTime.utc().setZone(action.config.timezone)
+        const nextTime = this.getNextValidTime(now, action.config)
+
+        return Promise.resolve({
+            finished: true,
+            scheduledAt: nextTime,
+        })
+    }
+
+    private getNextValidTime(now: DateTime, config: Action['config']): DateTime {
+        // If time is 'any', just find next valid day
+        if (config.time === 'any') {
+            return this.getNextValidDay(now, config.date)
+        }
+
+        const [startTime, endTime] = config.time
+        const [startHours, startMinutes] = startTime.split(':').map(Number)
+        const [endHours, endMinutes] = endTime.split(':').map(Number)
+
+        // Try today first
+        let nextTime = now.set({ hour: startHours, minute: startMinutes, second: 0, millisecond: 0 })
+        const endTimeToday = now.set({ hour: endHours, minute: endMinutes, second: 0, millisecond: 0 })
+
+        // If we're within the time window today, schedule for end of window
+        if (now >= nextTime && now <= endTimeToday && this.isValidDay(now, config.date)) {
+            return DateTime.now().setZone(config.timezone)
path: plugin-server/src/cdp/services/hogflows/actions/action.wait_until_time_window.ts
text: @user1:
logic: Using DateTime.now() here instead of the passed 'now' parameter introduces inconsistency. Should use the same time reference throughout.

```suggestion
            return now
```