Add a channels filter to the checkpoint list endpoint

Eduardo Bonetrequested to merge
631230-add-a-channels-filter-to-the-duo-workflow-che into master

What does this MR do and why?

GET /api/v4/ai/duo_workflows/workflows/:id/checkpoints now accepts an optional channels query param that returns only the requested channels in each checkpoint's channel_values. The AI Gateway's stop-recovery walk pages this endpoint but reads only the status channel of the checkpoints it walks past, and the endpoint used to rebuild every channel of every checkpoint on the page. Callers that omit channels get the same response, queries, and query count as before.

Design notes
  • ee/app/models/ai/duo_workflows/workflow.rb: blobs_by_thread_ts_for(checkpoints, channels:) filters the blob query on the requested channels intersected with the union of the page headers' channel_keys. When any header on the page has no channel_keys, or the union has more than 100 entries, it filters on the requested channels alone. This reuses the channel IN (...) clause that the membership filter already adds in !252495 (merged), so the query shape stays the same, and the list stays at most 100 items because the param is capped at 100.
  • reconstructed_channel_values_from(checkpoint, blobs, channels:) slices the header's stored channel_values to the requested channels before the early return for "no blobs", so every return path is filtered.
  • group_bounded_blobs_from now requires a compaction anchor only for the requested channels (the header's membership intersected with channels). Without this change, a channels=status request would refetch the whole chain whenever an unrequested channel lacked its anchor.
  • unbounded_blobs_by_thread_ts, the refetch used when a group lacks a compaction anchor, is filtered by the requested channels too, and memoized per channel list, so a page still refetches at most once.
  • ee/lib/api/ai/duo_workflows/workflows_internal.rb: on the legacy branch (dw_read_blobs_list off, rows read from p_duo_workflows_checkpoints), channel_values lives in one jsonb column, so the endpoint slices it after loading.
  • The param is Array[String], coerced with API::Validations::Types::CommaSeparatedToArray, so both channels=status,ui_chat_log and channels[]=status work. An empty list returns 400. The list allows at most 100 items, each 1 to 255 characters. A repeated key without brackets (channels=a&channels=b) keeps only the last value, which is standard Rack behavior. Unknown channel names are dropped without an error.
  • checkpoint_writes and channel_versions are not filtered; only channel_values is.
  • doc/api/openapi/openapi_v3.yaml is regenerated for the new param.
  • The current AI Gateway does not send channels, so it sees no change. The gateway MR gitlab-org/modelops/applied-ml/code-suggestions/ai-assist!7141 (merged) sends channels[]=status, and a request spec on this branch covers that form. A Rails version without this MR ignores the param, because Grape ignores undeclared params, and returns full checkpoints, so either deploy order is safe.
  • On a stop-recovery RESUME, the gateway also needs the boundary checkpoint's ui_chat_log, because it is the baseline for the cancelled-turn context. The gateway MR fetches the boundary in full through by_thread_ts, and LangGraph loads it again to resume. Per resume with the boundary on the first page, the local estimate is about 436 MB today, about 53 MB with both MRs (about 88% less), and about 28 MB if the gateway reuses its boundary fetch for LangGraph's load.
  • !258386 (merged) edits the same reconstruct_page! and reconstructed_channel_values_from. Whichever MR merges second needs a rebase.
Benchmark

Local GDK, synthetic workflow with 60 checkpoints in one current_thread group, 30 KB and 15 KB per-step deltas, compressed responses, MemoryProfiler total allocated bytes (the same kind of metric as mem_total_bytes in the API logs), second run of each. Not production data.

Request Allocated
per_page=20 410 MB
per_page=20, channels=status 1.6 MB
per_page=20, channels=status,ui_chat_log 123 MB
by_thread_ts for the boundary checkpoint 26 MB

References

Screenshots or screen recordings

No UI change.

How to set up and validate locally

  1. Check out this branch and open a Rails console: bin/rails console.
  2. Create a workflow with one checkpoint holding two channels: 
    user = User.find_by_username('root')
workflow = Ai::DuoWorkflows::Workflow.create!(user: user, project: Project.first, goal: 'channels QA',
  incremental_checkpoints_enabled: true, environment: :web)
blob = ->(channel, value) do
  { channel: channel, version: '1', write_type: 'json', step_action: 'compaction',
    data: Base64.strict_encode64(Zlib::Deflate.deflate(value.to_json)) }
end
ts = Gitlab::Utils.uuid_v7
Ai::DuoWorkflows::CreateCheckpointService.new(workflow: workflow, params: {
  thread_ts: ts, current_thread: 0,
  checkpoint: { 'id' => ts, 'ts' => Time.current.iso8601, 'channel_versions' => {}, 'channel_values' => {} },
  metadata: { 'step' => 0 }, channel_keys: %w[status ui_chat_log],
  channel_blobs: [blob.call('status', 'INPUT_REQUIRED'), blob.call('ui_chat_log', [{ 'content' => 'hi' }])]
}).execute
headers = workflow.checkpoint_headers.to_a
  3. Rebuild without a filter and confirm both channels come back: 
    workflow.reconstructed_channel_values_from(headers.first, workflow.blobs_by_thread_ts_for(headers))
    Expect: {"status"=>"INPUT_REQUIRED", "ui_chat_log"=>[{"content"=>"hi"}]}
  4. Rebuild with the filter the endpoint applies for ?channels=status: 
    workflow.reconstructed_channel_values_from(headers.first,
  workflow.blobs_by_thread_ts_for(headers, channels: %w[status]), channels: %w[status])
    Expect: {"status"=>"INPUT_REQUIRED"}
  5. Run the endpoint specs: 
    bin/rspec ee/spec/requests/api/ai/duo_workflows/workflows_internal_spec.rb -e 'with channels parameter'
    Expect: all examples pass.
Edge cases

Each case below has a spec:

  • channels= (empty) returns 400.
  • A list of 101 channels returns 400.
  • A channel the header does not declare returns empty channel_values.
  • A group without a compaction anchor refetches with the channel filter applied (two blob queries, both filtered on status).
  • The legacy branch slices channel_values the same way.

Database

This MR narrows or adds a channel filter on two blob queries on p_duo_workflows_checkpoint_blobs, with no new index or migration. On a production workflow with 1,139 checkpoints, a channels=status read uses the existing (workflow_id, channel, id) index and reads 503 buffers, compared with 2,375 for the master page query and 1,482 for the master refetch.

Query plans
  • Two queries change: the page query (Workflow#blobs_by_thread_ts_for) and the refetch query (Workflow#unbounded_blobs_by_thread_ts), which runs when a group lacks a compaction anchor.
  • On master, the page query filters channel by the union of the page headers' channel_keys, or has no filter when a header lacks one. The refetch query has no channel filter.
  • With channels=status, both queries filter channel = 'status'. A multi-channel request such as channels=status,ui_chat_log produces channel IN (...), the same shape as the master page query below. Callers that omit channels run the master queries unchanged.
  • Plans come from Database Lab (gitlab-production-main), with data as of 2026-09-29 04:24 UTC. The plans shorten the thread_ts list for readability. The 1,139-value list already exists on master, and this MR does not change it. Planning takes 90 to 170 ms with it.
  • Test workflow 8711521 has the most checkpoints of the workflows created on 2026-09-27: 1,139 checkpoints in one current_thread group and 3,528 blobs. One group and a linear chain give both queries the same 1,139-value thread_ts list, so one channels=status plan covers both.
  • With channel = 'status', Postgres uses idx_duo_wf_checkpoint_blobs_on_workflow_channel_id and returns rows in id order without a sort: 506 rows, 503 buffers, 2.3 ms.
  • The master page query uses the same index plus a sort (2,427 rows, 2,375 buffers, 7.0 ms). The master refetch query uses idx_duo_wf_checkpoint_blobs_dedup plus a sort (3,528 rows, 1,482 buffers, 112.7 ms, most of it cold reads).

This MR: page query and refetch with channels=status

SELECT "p_duo_workflows_checkpoint_blobs".* FROM "p_duo_workflows_checkpoint_blobs" WHERE "p_duo_workflows_checkpoint_blobs"."workflow_id" = 8711521 AND "p_duo_workflows_checkpoint_blobs"."workflow_created_at" = '2026-09-27 13:05:42.680577' AND "p_duo_workflows_checkpoint_blobs"."project_id" = 84861201 AND "p_duo_workflows_checkpoint_blobs"."thread_ts" IN (… 1,139 thread_ts values …) AND "p_duo_workflows_checkpoint_blobs"."channel" = 'status' ORDER BY "p_duo_workflows_checkpoint_blobs"."id" ASC
 Index Scan using index_22d319300b on gitlab_partitions_dynamic.p_duo_workflows_checkpoint_blobs_20260927 p_duo_workflows_checkpoint_blobs  (cost=3.27..270.06 rows=1 width=369) (actual time=0.107..2.257 rows=506 loops=1)
   Index Cond: ((p_duo_workflows_checkpoint_blobs.workflow_id = 8711521) AND (p_duo_workflows_checkpoint_blobs.channel = 'status'::text))
   Filter: ((p_duo_workflows_checkpoint_blobs.workflow_created_at = '2026-09-27 13:05:42.680577+00'::timestamp with time zone) AND (p_duo_workflows_checkpoint_blobs.project_id = 84861201) AND (p_duo_workflows_checkpoint_blobs.thread_ts = ANY ('{… 1,139 thread_ts values …}'::text[])))
   Buffers: shared hit=503
   I/O Timings: read=0.000 write=0.000
Settings: random_page_cost = '1.5', seq_page_cost = '4', effective_cache_size = '472585MB', work_mem = '230MB', jit = 'off'
Time: 94.258 ms
  - planning: 91.937 ms
  - execution: 2.321 ms

Master: page query (header channel_keys filter)

SELECT "p_duo_workflows_checkpoint_blobs".* FROM "p_duo_workflows_checkpoint_blobs" WHERE "p_duo_workflows_checkpoint_blobs"."workflow_id" = 8711521 AND "p_duo_workflows_checkpoint_blobs"."workflow_created_at" = '2026-09-27 13:05:42.680577' AND "p_duo_workflows_checkpoint_blobs"."project_id" = 84861201 AND "p_duo_workflows_checkpoint_blobs"."thread_ts" IN (… 1,139 thread_ts values …) AND "p_duo_workflows_checkpoint_blobs"."channel" IN ('status', 'conversation_history', 'ui_chat_log', 'context', 'agent_context_limits', 'branch:to:user_input#fetch') ORDER BY "p_duo_workflows_checkpoint_blobs"."id" ASC
 Sort  (cost=1747.74..1747.75 rows=1 width=369) (actual time=6.602..6.829 rows=2427 loops=1)
   Sort Key: p_duo_workflows_checkpoint_blobs.id
   Sort Method: quicksort  Memory: 1774kB
   Buffers: shared hit=2375
   I/O Timings: read=0.000 write=0.000
   ->  Index Scan using index_22d319300b on gitlab_partitions_dynamic.p_duo_workflows_checkpoint_blobs_20260927 p_duo_workflows_checkpoint_blobs  (cost=3.27..1747.73 rows=1 width=369) (actual time=0.093..5.171 rows=2427 loops=1)
         Index Cond: ((p_duo_workflows_checkpoint_blobs.workflow_id = 8711521) AND (p_duo_workflows_checkpoint_blobs.channel = ANY ('{status,conversation_history,ui_chat_log,context,agent_context_limits,branch:to:user_input#fetch}'::text[])))
         Filter: ((p_duo_workflows_checkpoint_blobs.workflow_created_at = '2026-09-27 13:05:42.680577+00'::timestamp with time zone) AND (p_duo_workflows_checkpoint_blobs.project_id = 84861201) AND (p_duo_workflows_checkpoint_blobs.thread_ts = ANY ('{… 1,139 thread_ts values …}'::text[])))
         Buffers: shared hit=2372
         I/O Timings: read=0.000 write=0.000
Settings: random_page_cost = '1.5', seq_page_cost = '4', effective_cache_size = '472585MB', work_mem = '230MB', jit = 'off'
Time: 178.330 ms
  - planning: 171.326 ms
  - execution: 7.004 ms

Master: refetch query (no channel filter)

SELECT "p_duo_workflows_checkpoint_blobs".* FROM "p_duo_workflows_checkpoint_blobs" WHERE "p_duo_workflows_checkpoint_blobs"."workflow_id" = 8711521 AND "p_duo_workflows_checkpoint_blobs"."workflow_created_at" = '2026-09-27 13:05:42.680577' AND "p_duo_workflows_checkpoint_blobs"."project_id" = 84861201 AND "p_duo_workflows_checkpoint_blobs"."thread_ts" IN (… 1,139 thread_ts values …) ORDER BY "p_duo_workflows_checkpoint_blobs"."id" ASC
 Sort  (cost=2296.56..2296.56 rows=1 width=369) (actual time=111.916..112.444 rows=3528 loops=1)
   Sort Key: p_duo_workflows_checkpoint_blobs.id
   Sort Method: quicksort  Memory: 1977kB
   Buffers: shared hit=1378 read=104
   I/O Timings: read=103.106 write=0.000
   ->  Index Scan using p_duo_workflows_checkpoint_b_project_id_workflow_id_threa_idx33 on gitlab_partitions_dynamic.p_duo_workflows_checkpoint_blobs_20260927 p_duo_workflows_checkpoint_blobs  (cost=0.55..2296.55 rows=1 width=369) (actual time=6.502..109.808 rows=3528 loops=1)
         Index Cond: ((p_duo_workflows_checkpoint_blobs.project_id = 84861201) AND (p_duo_workflows_checkpoint_blobs.workflow_id = 8711521) AND (p_duo_workflows_checkpoint_blobs.thread_ts = ANY ('{… 1,139 thread_ts values …}'::text[])) AND (p_duo_workflows_checkpoint_blobs.workflow_created_at = '2026-09-27 13:05:42.680577+00'::timestamp with time zone))
         Buffers: shared hit=1375 read=104
         I/O Timings: read=103.106 write=0.000
Settings: jit = 'off', random_page_cost = '1.5', seq_page_cost = '4', effective_cache_size = '472585MB', work_mem = '230MB'
Time: 202.610 ms
  - planning: 89.930 ms
  - execution: 112.680 ms

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 Eduardo Bonet

Merge request reports