Enable SAs to run system-triggered Duo flows without composite identity

What does this MR do and why?

This MR completes: #605895 and is a follow up to the DB setup work: Add trigger metadata columns to duo_workflows_w... (!247767 - merged)

Note: This work aims to unblock the CRON / scheduled trigger work: #594180 and is intended as an experimental V1, so we can move quickly and get scheduled triggers into internal teams' hands for dogfooding.

Summary

Enables group/project-scoped service accounts to execute system-triggered Duo workflows without composite identity, acting as both workflow owner and runtime identity.

What

Adds an autonomous execution path in FlowTriggers::RunService that allows SAs to run flows when triggered by system events (e.g., upcoming scheduled trigger work) without requiring a human initiator or composite identity linking.

Why

Today, all flow trigger execution requires a human to initiate the action and link their identity to the SA via composite identity. This blocks system-triggered scenarios where no human is present, such as scheduled workflows, etc. These use cases need the SA to execute autonomously while maintaining security boundaries.

How it works

• RunService accepts a trigger_source parameter (:human default, :system, or :scheduled). When :system/:scheduled, the autonomous path is used, with composite ID linking is skipped since there's no current_user, and a narrower OAuth token is issued without user:$ID scopes.
• AutonomousServiceAccountEligibilityValidator enforces that the SA is a service account, group/project-scoped (not instance-wide), active, and a project member before every autonomous run to ensure security requirements are met as per point 1 and 4 of implementation design.
• CreateAutonomousOauthAccessTokenService issues credentials scoped to ai_workflows only, with the SA as resource owner and 1-hour expiry as per security requirement (point 6 of implementation design).
• trigger_source enum on the Workflow model (attr_readonly) provides immutable attribution so downstream systems can distinguish autonomous runs from human-triggered runs.
• The same SA provisioned during flow enablement is reused, the only difference is runtime behavior based on trigger_source.

How future system triggers will use this

Workers that handle system events can use the autonomous path by passing trigger_source: :system or :scheduled with no current_user. For example:

Ai::FlowTriggers::RunService.new(
    project: project,
    flow_trigger: trigger,
    trigger_source: :system
).execute(params)

Verification Scenarios

#	Scenario	Expected	Verified
1	Autonomous path with flow-provisioned SA (trigger_source: :scheduled, group-scoped, active, project member)	autonomous_trigger? returns true, validation passes	✅
2	Composite identity path with same SA and trigger human user (trigger_source: :human)	autonomous_trigger? returns false, can_use_composite_identity? returns true, validation passes	✅
3	Autonomous token scopes	["ai_workflows"] only, no user:$ID, SA as resource owner, 1hr (3600s) expiry	✅
4	Human path regression - human user still passes (trigger_source: :human)	autonomous_trigger? returns false, validation passes	✅
5	SA on human path rejected (current_user is SA, trigger_source: :human)	"cannot be triggered by non-human users"	✅
6	Instance-wide SA rejected	"Service account must be group or project scoped, not instance-wide"	✅
7	Blocked SA rejected	"Service account must be active"	✅
8	Non-member SA rejected	"Service account must be a member of the target project"	✅

Verify locally

Pre-requisites:

• Duo Enterprise enabled
• GDK running with all services up

1. Checkout this branch 605895/sq/enables-sa-without-composite-id

2. Enable feature flag in rails console: Feature.enable(:autonomous_service_account_execution)

3. In GitLab UI:

   1. Step 1: Enable or create a flow at the project level via UI:
      • Navigate to http://localhost:3000/test-status-bot-e63fd58e/test/-/settings/duo_agent_platform
      • Enable or create a flow (e.g., a foundational flow or any available flow/agent)
      • This auto-provisions a group-level SA and creates the item consumer chain
      • Once you've enabled a flow, run the following scripts below in the Rails console:

4. In Rails Console (open details below for scripts):

# Get your project and user
project = Project.find(1000000)
human_user = User.find_by(username: 'root')

# Find the most recently created item consumer for this project (i.e. flow created in Step 1)
consumer = Ai::Catalog::ItemConsumer.where(project: project).order(created_at: :desc).first

if consumer
  sa = consumer.active_service_account
  puts "Consumer: #{consumer.id} (item: #{consumer.item.name})"
  puts "SA: #{sa&.username} (id: #{sa&.id})"
  puts "  provisioned_by_group_id: #{sa&.provisioned_by_group_id}"
  puts "  composite_identity_enforced: #{sa&.composite_identity_enforced?}" # true
  puts "  active: #{sa&.active?}" # true
  puts "  project member: #{project.member?(sa)}" # true
  puts "Parent consumer: #{consumer.parent_item_consumer&.id}"
else
  puts "No item consumer found for this project"
end


# Create a trigger for this consumer to verify SA provisioned through flow enablement is group-scoped, composite identity, active, project member (existing behaviour preserved):
trigger = Ai::FlowTrigger.create!(
  project: project,
  ai_catalog_item_consumer: consumer,
  description: 'Verification trigger for sunny-thurs-flow',
  event_types: [Ai::FlowTrigger::EVENT_TYPES[:merge_request]]
)

puts "Trigger: #{trigger.id}"
puts "Trigger SA: #{trigger.service_account.username} (id: #{trigger.service_account.id})" # inherits SA from recently created consumer in Step 1

# Test the autonomous path with flow-provisioned SA
service = Ai::FlowTriggers::RunService.new(
  project: project,
  flow_trigger: trigger,
  trigger_source: :scheduled
)

puts "autonomous_trigger?: #{service.send(:autonomous_trigger?)}" # true
error = service.send(:validation_error)
puts "Autonomous validation: #{error&.message || 'PASSED'}" # PASSED

#  Now test composite identity still works with the same SA and trigger when trigger_source is human (existing behaviour preserved):
service_composite = Ai::FlowTriggers::RunService.new(
  project: project,
  flow_trigger: trigger,
  current_user: human_user,
  trigger_source: :human
)

puts "autonomous_trigger?: #{service_composite.send(:autonomous_trigger?)}" # false
puts "can_use_composite_identity?: #{service_composite.send(:can_use_composite_identity?)}" # true
error = service_composite.send(:validation_error)
puts "Composite identity validation: #{error&.message || 'PASSED'}" # PASSED

# Autonomous token has narrow scopes (ai_workflows), no user:$ID, SA is the owner, 1-hour expiry.
result = Ai::DuoWorkflows::CreateAutonomousOauthAccessTokenService.new(
  service_account: sa,
  organization: project.organization,
  container: project
).execute

puts "Token created: #{result.success?}" # true
if result.success?
  token = result[:oauth_access_token]
  puts "  Scopes: #{token.scopes.to_a}" # ["ai_workflows"]
  puts "  Has user:ID scope: #{token.scopes.to_a.any? { |s| s.start_with?('user:') }}" # false
  puts "  Owner: #{token.resource_owner_id} (SA: #{sa.id})" # same ID as the SA, e.g. 48 (SA: 48)
  puts "  Expires in: #{token.expires_in}s" # 3600s
end

# Human path / trigger source works as expected (existing behaviour)
service_human = Ai::FlowTriggers::RunService.new(
  project: project,
  flow_trigger: trigger,
  current_user: human_user,
  trigger_source: :human
)

puts "Human path autonomous?: #{service_human.send(:autonomous_trigger?)}" # false
error = service_human.send(:validation_error) # nil
puts "Human user validation: #{error&.message || 'PASSED'}" # PASSED

# Autonomous SA on the human path is rejected
service_sa_human = Ai::FlowTriggers::RunService.new(
  project: project,
  flow_trigger: trigger,
  current_user: sa,
  trigger_source: :human
)

error = service_sa_human.send(:validation_error)
puts "SA on human path: #{error&.message}" # cannot be triggered by non-human users

# Instance-wide SA rejected (not eligible)
admin = User.find_by(username: 'root')
result = Users::ServiceAccounts::CreateService.new(admin).execute
instance_sa = result.payload[:user]

puts "ID: #{instance_sa.id}"
puts "provisioned_by_group_id: #{instance_sa.provisioned_by_group_id}"
puts "provisioned_by_project_id: #{instance_sa.provisioned_by_project_id}"

validator =
Ai::FlowTriggers::AutonomousServiceAccountEligibilityValidator.new(instance_sa, project)
puts "Eligible: #{validator.valid?}" # false
puts "Errors: #{validator.errors.full_messages}" # Errors: ["Service account must be group or project scoped, not instance-wide", "Service account must be a member of the target project"]

# Blocked SA doesn't work (not eligible)
sa.block!
validator = Ai::FlowTriggers::AutonomousServiceAccountEligibilityValidator.new(sa, project)
puts "Blocked SA eligible: #{validator.valid?}" # false
puts "Errors: #{validator.errors.full_messages}" # Errors: ["Service account must be active"]
sa.activate!
puts "SA reactivated"

# Non project member SA doesn't work (not eligible)

# Create a group-scoped SA in a different group that has no access to this project
other_group = Group.where.not(id: project.namespace_id).first
admin = User.find_by(username: 'root')

result = Namespaces::ServiceAccounts::GroupCreateService.new(
  admin, { organization_id: other_group.organization_id, namespace_id: other_group.id }
).execute

non_member_sa = result.payload[:user]

puts "SA: #{non_member_sa.username}"
puts "Provisioned by group: #{non_member_sa.provisioned_by_group_id} (project group:
#{project.namespace_id})"
puts "Project member: #{project.member?(non_member_sa)}"

Gitlab::SafeRequestStore.clear!
validator = Ai::FlowTriggers::AutonomousServiceAccountEligibilityValidator.new(non_member_sa, project)
puts "Eligible: #{validator.valid?}" # false
puts "Errors: #{validator.errors.full_messages}" # Errors: ["Service account must be a member of the target project"]

Related

1. Add trigger metadata columns to duo_workflows_w... (!247767 - merged)
2. Implement trigger: Scheduled / Cron Trigger (#594180)

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.