Adds flow schedule execution

Shola Quadrirequested to merge
594180/sq/flow-schedule-execution into master

What does this MR do and why?

MR 6 (of 7) that addressses: #594180. The goal of the series: users will be able to run AI flows on a cron schedule.

This MR adds flow schedules execution.

The earlier MRs in the stack lets us create an Ai::FlowSchedule row with a next_run_at, but nothing ever reads or acts on it until now. This MR is what makes scheduling functionally real by introducing a dispatcher that watches for due schedules and a worker that actually runs them.

  • FlowScheduleWorker: cron dispatcher on a jittered cron (3-59/10 * * * *, matching the pipeline schedule load flattening pattern), gated by an instance-wide flag, and uses batching of 500 at a time with a 7s stagger as per implementation design
  • RunFlowScheduleWorker: idempotent, deduplicated executor that advances next_run_at before executing (drops on failure; a failed tick is skipped, never retried) and records success/failure with 3-strike deactivation
  • RunService gains a flow_schedule: param and threads trigger metadata (trigger_source, trigger/schedule ids) through the catalog execution chain which is set at the workflow creation since trigger_source is attr_readonly
  • We also add an autonomous OAuth token branch in WorkflowContextGenerationService since SA acts alone for autonomous triggers with no user scope

How to set up and validate locally

Prerequisites

  • Rails console
  • GDK and letter_opener running: http://172.16.123.1:3000/rails/letter_opener
  1. Set up in Rails (create project, users, trigger and schedule):

1. Project, SA, membership and owner set up

# fetch or create project
project = Project.find(1000000)

# create SA
sa = begin
  FactoryBot.create(:service_account,
    username: "sa_#{SecureRandom.hex(6)}", email: "sa_#{SecureRandom.hex(6)}@example.com",
    provisioned_by_group: begin
      FactoryBot.create(:group, path: "group-#{SecureRandom.hex(6)}")
    rescue ActiveRecord::RecordInvalid
      retry
    end)
rescue ActiveRecord::RecordInvalid
  retry
end

# Add SA to project
project.add_developer(sa)
sa.refresh_authorized_projects(source: :test)
project.member?(sa) # => true

# Create owner and add to project
owner = project.owners_and_maintainers.first
if owner.nil?
  owner = begin
    FactoryBot.create(:user, username: "owner_#{SecureRandom.hex(6)}", email: "owner_#{SecureRandom.hex(6)}@example.com")
  rescue ActiveRecord::RecordInvalid
    retry
  end
  project.add_owner(owner)
end

2. FF and Duo set up

# enable feature flags
Feature.enable(:ai_flow_schedules) 
Feature.enable(:autonomous_service_account_execution, project) # needed for :scheduled event type + RunService's composite-identity bypass

Gitlab::Llm::StageCheck.define_singleton_method(:available?) { |*| true }

# Class-level, not instance-level: the real run loads fresh User records deep inside
# the worker, so a singleton override on a local variable never applies to those.
User.class_eval do
  def allowed_to_use?(*)
    true
  end
end

3. Trigger and schedule set up

# Create scheduled trigger
trigger = Ai::FlowTrigger.create!(
  project: project,
  user: sa,
  event_types: [Ai::FlowTrigger::EVENT_TYPES[:scheduled]],
  description: 'Nightly run',
  config_path: 'duo/does-not-exist.yml',   # guarantees a real failure on first run
  filter: { 'scheduled' => { 'frequency' => 'DAILY', 'minute' => 30, 'hour' => 14 } }
)

# Fire trigger
Ai::FlowSchedules::SyncFromTriggerService.new(trigger: trigger).execute
schedule = trigger.flow_schedules.first
schedule.update_column(:next_run_at, 1.minute.ago)

4. Run and verify failure path (run everything in the same block)

result = Ai::FlowSchedule.find(schedule.id)
{
  fresh: (Time.current - result.last_run_at < 30),
  last_run_status: result.last_run_status,
  last_run_error: result.last_run_error,
  consecutive_failure_count: result.consecutive_failure_count,
  active: result.active
}

# Expect fresh: true, last_run_error: "invalid or missing flow definition", consecutive_failure_count: 1. 

5. Drive to 3-strike deactivation

2.times do
  Ai::RunFlowScheduleWorker.new.perform(schedule.id, {})
end

result = Ai::FlowSchedule.find(schedule.id)
{
  fresh: (Time.current - result.last_run_at < 30),
  consecutive_failure_count: result.consecutive_failure_count,
  active: result.active
}

Expect fresh: true, consecutive_failure_count: 3, active: false. 

6. Confirm the deactivation notification email in http://localhost:3000/rails/letter_opener

7. Success / happy path: fix the flow config, reset and rerun in the same block

trigger.update!(config_path: 'duo/flow.yml')

content = "version: v1\nenvironment: ambient\nimage: alpine:latest\ncommands:\n  - echo hello\ncomponents: []\nrouters: []\nflow: []\n"
unless project.repository.blob_data_at(project.repository.root_ref, 'duo/flow.yml')
  project.repository.create_file(sa, 'duo/flow.yml', content,
    message: 'Add flow config', branch_name: project.repository.root_ref)
end

fresh_schedule = Ai::FlowSchedule.find(schedule.id)
fresh_schedule.update!(active: true, consecutive_failure_count: 0, next_run_at: 1.minute.ago)

Ai::RunFlowScheduleWorker.new.perform(fresh_schedule.id, { 'scheduling' => true })

result = Ai::FlowSchedule.find(schedule.id)
{
  fresh: (Time.current - result.last_run_at < 30),
  last_run_status: result.last_run_status,
  last_run_error: result.last_run_error
}

# Expect fresh: true, last_run_status: "success", last_run_error: nil. 

8. Clean up

trigger.destroy # cascades to its flow_schedules

Related Stacked MRs (in order)

  1. Adds ai_flow_schedules table for scheduled AI f... (!250227 - merged)
  2. Add scheduled event type to AI flow triggers (!250805 - merged)
  3. Adds Flow Schedule Model (!250809 - merged)
  4. Add Flow Schedules API (!250821 - merged)
  5. Adds Flow Schedule Email Notifications (!250824 - merged)
  6. Adds flow schedule execution (!250838) (This MR)
  7. Adds FE changes for Flow Schedules (!250843)

DB Query

Up to date query

SELECT ai_flow_schedules.id, ai_flow_schedules.ai_flow_trigger_id, ai_flow_schedules.project_id 
FROM ai_flow_schedules 
WHERE active = TRUE AND next_run_at < now() 
  AND id > ( SELECT id FROM ai_flow_schedules WHERE active = TRUE AND next_run_at < now() ORDER BY id ASC LIMIT 1 OFFSET 499 ) 
ORDER BY id ASC LIMIT 500;

Old query (ignore):

SELECT "ai_flow_schedules".*
FROM "ai_flow_schedules"
WHERE "ai_flow_schedules"."active" = TRUE
  AND "ai_flow_schedules"."next_run_at" < now();

Query Execution Plan

References

Screenshots or screen recordings

Before After

How to set up and validate locally

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.

Edited by Shola Quadri

Merge request reports