Bound chains of workflows started by other workflows

Igor Drozdovrequested to merge
630476-flow-trigger-chaining-limits into master

What does this MR do and why?

This is MR 2 of the plan to let AI flows trigger other flows, with limits. It lets CreateWorkflowService accept a parent workflow and checks limits before it creates the child. It also records the trigger and event for every flow trigger run.

This MR builds on !258704 (merged) (MR 1), which is merged. MR 1 added the parent_workflow_id column, the flow trigger source, and the parent_workflow / child_workflows associations.

No caller passes a parent yet, so the chaining part does not change anything for users. The agent_workflows API path (MR 3) and the mention path (MR 4) will pass a parent later. All parent handling is behind the new wip feature flag ai_flow_trigger_chaining. It is off by default. The actor is the container's top-level namespace. When the flag is off, a passed parent is ignored.

Changes in commit 1, "Bound chains of workflows started by other workflows":

  • Add Ai::FlowTriggers::CallerContext in ee/app/services/ai/flow_triggers/caller_context.rb. Given a parent workflow, it works out:
    • the depth of the new child, and whether the depth limit is hit
    • whether the parent's breadth limit is hit
    • the human who started the chain (root_human)
    • whether a container is in the same top-level namespace as the parent
  • Limits are MAX_TRIGGER_DEPTH = 3 and MAX_CHILD_WORKFLOWS_PER_PARENT = 5. One human action can lead to at most 30 flow-triggered sessions.
  • Depth: a workflow with no parent has depth 0. A(0) -> B(1) -> A(2) is allowed. The next hop, B(3), is rejected. Verification workflows are not counted.
  • The walk up the chain loads at most MAX_TRIGGER_DEPTH + 1 workflows, with one primary key lookup each. It never loads a whole long chain.
  • root_human is nil when:
    • the root is autonomous (system or scheduled)
    • the root user is not human
    • the root was itself started by a flow, but its parent was deleted. This makes sure that deleting ancestors cannot reset the depth count.
  • Breadth counts every child of the parent, whatever its status.
  • Ai::DuoWorkflows::CreateWorkflowService accepts a parent_workflow param. When the flag is on and a parent is given, it sets trigger_source: :flow and the parent. It rejects the workflow with HTTP status :forbidden and one of these reasons:
    • :invalid_trigger_source when the caller also passed a trigger source other than flow, for example system or verification. Those change how the run is authorized and billed.
    • :parent_workflow_namespace_mismatch when the parent is in a different top-level namespace. MR 1 listed a same project or namespace check as a follow-up. This MR uses the top-level namespace, so a flow can hand off to a sibling project in the same group.
    • :trigger_depth_exceeded
    • :chain_root_not_human
    • :trigger_breadth_exceeded
  • The breadth check and the save run inside with_lock('FOR NO KEY UPDATE') on the parent. This stops parallel requests from going past the limit. FOR NO KEY UPDATE still lets the running parent insert rows that point to it, such as checkpoints and events.
  • parent_workflow and parent_workflow_id are removed from the generic params. The parent can only be set through the checked path. No REST or GraphQL endpoint accepts these params from clients.
  • Ai::Catalog::Flows::ExecuteService and Ai::Catalog::ExecuteWorkflowService pass parent_workflow through to CreateWorkflowService.
  • Ai::DuoWorkflows::RestartWorkflowService passes the original workflow's parent. A restarted child stays in its chain and counts against the limits. It no longer copies trigger_source: flow, because CreateWorkflowService sets it when it attaches the parent. When the parent was deleted or the flag is off, the restart runs as a normal session outside any chain. This resolves the restart follow-up listed in MR 1.

Changes in commit 2, "Record the trigger and event for every flow trigger run":

  • Ai::FlowTriggers::RunService now writes trigger_flow_trigger_id and trigger_event_type for every run, not only autonomous ones. !255769 (merged) already writes both for autonomous runs. trigger_source and trigger_flow_schedule_id are still only written for autonomous runs.
  • It reuses the resolved_trigger_event_type helper from that MR, so unknown event names are still ignored.
  • ExecuteWorkflowService builds the trigger values in one trigger_metadata_params method.
  • This part is not behind the flag. One visible effect: RestartWorkflowService reads trigger_event_type and uses api_execution when it is empty. Restarts of runs that a human triggered now use the original event, for example mention or assign, as the restart service intended.

Follow-ups

These come in later MRs in the plan:

  • MR 3: the agent_workflows API path.
  • MR 4: the RunService guard, root human permission checks, and composite identity for the mention path.
  • MR 5: logging, internal events, and docs.

There is no rollout issue yet. The flag is wip and will get a rollout issue before it is turned on.

Database

No schema changes. New queries, which only run when a parent is passed and the flag is on:

  • One primary key lookup per ancestor, at most 4.
  • SELECT ... FOR NO KEY UPDATE on the parent row.
  • COUNT(*) of children, using index_duo_workflows_workflows_on_parent_workflow_id from MR 1.

References

Screenshots or screen recordings

Not applicable. This is a backend only change.

How to set up and validate locally

  1. Check out the branch and run bin/rails db:migrate.

  2. In bin/rails console, create a child workflow:

    Feature.enable(:ai_flow_trigger_chaining)
parent = Ai::DuoWorkflows::Workflow.where(trigger_source: :human).last
result = Ai::DuoWorkflows::CreateWorkflowService.new(
  container: parent.project, current_user: parent.user,
  params: { environment: 'ide', goal: 'test', parent_workflow: parent }
).execute
result.payload[:workflow].parent_workflow == parent # => true
result.payload[:workflow].triggered_by_flow?        # => true

  3. Check the depth limit. Keep adding children until the chain has 3 workflows, then pass the last one as the parent:

    create = ->(p) { Ai::DuoWorkflows::CreateWorkflowService.new(container: p.project, current_user: p.user, params: { environment: 'ide', goal: 'test', parent_workflow: p }).execute }
workflow = create.(result.payload[:workflow]).payload[:workflow]
Ai::FlowTriggers::CallerContext.new(parent_workflow: workflow).depth # => 3
create.(workflow).payload[:reason] # => :trigger_depth_exceeded

  4. Run the specs:

    bin/rspec ee/spec/services/ai/flow_triggers/caller_context_spec.rb \
  ee/spec/services/ai/duo_workflows/create_workflow_service_spec.rb \
  ee/spec/services/ai/duo_workflows/restart_workflow_service_spec.rb \
  ee/spec/services/ai/flow_triggers/run_service_spec.rb \
  ee/spec/services/ai/catalog/flows/execute_service_spec.rb \
  ee/spec/services/ai/catalog/execute_workflow_service_spec.rb

MR acceptance checklist

Evaluate this MR against the checklist at https://docs.gitlab.com/development/code_review/#acceptance-checklist.

Edited by Igor Drozdov

Merge request reports