Reconstruct internal checkpoint list from blobs when incremental-only

What does this MR do?

Fixes #612674 (closed).

GET .../workflows/:id/checkpoints (ee/lib/api/ai/duo_workflows/workflows_internal.rb) read only the legacy p_duo_workflows_checkpoints table. Once duo_workflow_write_incremental_only is on, CreateCheckpointService stops writing that row -- only a checkpoint_headers row (+ blobs) is written -- so checkpoints written while the flag is on silently vanished from this list.

Two Duo Workflow Service consumers depend on this list staying complete:

• checkpoints_reversed() -- the STOP_RECOVERY boundary walk.
• _fetch_most_recent_checkpoint() -- the pre-18.8 GraphQL fallback.

Same root cause and fix pattern as !249802 (merged) (GraphQL latestCheckpoint/firstCheckpoint). Behind a new dw_read_blobs_list flag, the endpoint reads checkpoint_headers and rebuilds channel_values from blobs.

An incremental workflow writes a header for every checkpoint -- CreateCheckpointService#write_checkpoint_header is gated on incremental_checkpoints_enabled? alone, independent of write_incremental_only (which only drops the full row). Since the read gate also requires incremental_checkpoints_enabled?, the headers table is already a superset of checkpoints on this path, so it reads headers alone rather than merging the two sources. That matters for correctness as well as simplicity: compaction truncates ui_chat_log in the full row, so for a dual-written checkpoint the blobs are the only full-history source. Reading one relation also keeps pagination in the database.

This matches every sibling consumer -- latest_checkpoint/first_checkpoint (!249802 (merged)), by_thread_ts (!247134 (merged)), and full_trace_channel_values all switch outright rather than merge.

Both of the page's extra reads are batched into one query each, and scoped to the current page rather than the whole result set:

• checkpoint_writes via CheckpointWrite.for_workflows_and_threads, since CheckpointHeader#checkpoint_writes is a plain query method rather than an eager-loadable association (BulkInsertSafe forbids the autosave association).
• blobs via Workflow#blobs_by_thread_ts_for, which unions the page's ancestor chains into one query; #reconstructed_channel_values_from then folds each checkpoint against that in-memory set. Chains overlap heavily along a lineage, so reconstructing per checkpoint previously meant N queries over largely the same rows.

Out of scope

GET .../checkpoints/:checkpoint_id (keyed by the legacy integer id) is intentionally untouched. There's no integer id for a header-only checkpoint to look up by once the legacy row stops being written. Its consumer (aget_tuple with a checkpoint_id) is being migrated to a new by_thread_ts route instead, in the separate !247133 (merged) (merged, read engine) / !247134 (merged) (open) chain.

Feature flag

dw_read_blobs_list (wip, default off). Rollout issue: #612821. Kept separate from dw_read_blobs_api (!247134 (merged)): that flag's whole design is "the gateway owns the read decision" -- the by_thread_ts endpoint doesn't re-check it, it just serves whenever a header exists, because the gateway only calls that route once it has decided to. This list endpoint has no gateway-side opt-in (checkpoints_reversed() and the pre-18.8 fallback call it unconditionally), so Rails has to gate it itself, the same way dw_read_blobs_trace/dw_read_blobs_graphql already do for their own consumers.

Testing

• ee/spec/models/ai/duo_workflows/workflow_spec.rb: #reconstruct_from_blobs_for_list?.
• ee/spec/models/ai/duo_workflows/checkpoint_header_spec.rb: .in_reverse_checkpoint_order.
• ee/spec/requests/api/ai/duo_workflows/workflows_internal_spec.rb: header list with reconstructed channel_values and ordering; a dual-written checkpoint returning the full blob history rather than the compaction-truncated row; database-level pagination (X-Total/X-Per-Page); checkpoint_ns filtering; accept_compressed against reconstructed data; checkpoint_writes batched into one query regardless of header count; and the flag-off legacy path.

The pre-existing legacy-path specs now pin dw_read_blobs_list: false. They build Checkpoint rows through the factory, without the header a real incremental workflow always gets, and the flag is default-on in tests -- same reason the POST specs in that file already pin duo_workflow_write_incremental_only: false.

Database

Adds one index in db/migrate/20260812160000_add_workflow_id_thread_ts_index_to_duo_wf_checkpoint_headers.rb, and drops the standalone workflow_id index it makes redundant:

COLUMNS = %i[workflow_id thread_ts id]
add_concurrent_partitioned_index(:p_duo_workflows_checkpoint_headers, COLUMNS, name: NEW_INDEX_NAME)
remove_concurrent_partitioned_index_by_name(:p_duo_workflows_checkpoint_headers, OLD_INDEX_NAME)

workflow_id is a leading prefix of the new index, so it still serves plain workflow_id lookups and the foreign key's cascade delete. Same helper and add-then-drop shape as !247133 (merged)'s ReplacePDuoWorkflowsCheckpointBlobsUniqueIndex on the sibling blobs table. workflow_created_at is deliberately not in the index: it is the partition key, so it prunes rather than filters within a partition.

Why

Readers order a workflow's headers by thread_ts (the authoritative DWS sequence) with id breaking ties between re-sends. On workflow_id alone, each sorted the workflow's whole header set. Three readers benefit -- the list endpoint added here, plus latest_checkpoint_header/earliest_checkpoint_header and checkpoint_header_for from the already-merged read engine (!247133 (merged)).

Query and plans

The list endpoint's read (GDK, one workflow with 25 headers, default page size):

SELECT "p_duo_workflows_checkpoint_headers".* FROM "p_duo_workflows_checkpoint_headers"
WHERE "p_duo_workflows_checkpoint_headers"."workflow_id" = 1031
  AND "p_duo_workflows_checkpoint_headers"."workflow_created_at" = '2026-08-12 16:02:19.039544'
ORDER BY "p_duo_workflows_checkpoint_headers"."thread_ts" DESC,
         "p_duo_workflows_checkpoint_headers"."id" DESC
LIMIT 20 OFFSET 0

Before (sorts every header of the workflow, then discards all but the page):

Limit  (cost=2.17..2.17 rows=1 width=220) (actual time=0.016..0.017 rows=20 loops=1)
  Buffers: shared hit=12
  ->  Sort  (cost=2.17..2.17 rows=1 width=220) (actual time=0.016..0.016 rows=20 loops=1)
        Sort Key: thread_ts DESC, id DESC
        Sort Method: quicksort  Memory: 28kB
        ->  Index Scan using index_duo_wf_checkpoint_headers_on_workflow_id on p_duo_workflows_checkpoint_headers_20260812  (cost=0.14..2.16 rows=1 width=220) (actual time=0.003..0.005 rows=25 loops=1)
              Index Cond: (workflow_id = 1031)
              Filter: (workflow_created_at = '...'::timestamptz)
Execution Time: 0.023 ms

After (no sort; stops at the page size):

Limit  (cost=0.14..2.16 rows=1 width=220) (actual time=0.004..0.006 rows=20 loops=1)
  Buffers: shared hit=5
  ->  Index Scan Backward using index_duo_wf_checkpoint_headers_on_workflow_id_thread_ts_id on p_duo_workflows_checkpoint_headers_20260812  (cost=0.14..2.16 rows=1 width=220) (actual time=0.004..0.005 rows=20 loops=1)
        Index Cond: (workflow_id = 1032)
        Filter: (workflow_created_at = '...'::timestamptz)
Execution Time: 0.010 ms

Buffers drop 12 -> 5 because the scan stops after 20 rows instead of reading all 25 and sorting. The two already-merged readers also lose their sort:

latest_checkpoint_header  -> Index Scan Backward using index_..._workflow_id_thread_ts_id (rows=1)  Execution Time: 0.008 ms
checkpoint_header_for     -> Index Scan Backward using index_..._workflow_id_thread_ts_id
                               Index Cond: ((workflow_id = 1033) AND (thread_ts = 'h-07'))          Execution Time: 0.005 ms

checkpoint_header_for now matches thread_ts as an index condition rather than filtering after the scan.

Size and reversibility

p_duo_workflows_checkpoint_headers was introduced in 19.2 and is partitioned daily with 30-day retention, so it is far below the large-table thresholds. Every read is keyed by workflow_id + workflow_created_at and prunes to a single daily partition.

Migration output (up and down, main):

== 20260812160000 AddWorkflowIdThreadTsIndexToDuoWfCheckpointHeaders: migrating
-- add_index(:p_duo_workflows_checkpoint_headers, [:workflow_id, :thread_ts, :id], {:name=>"index_duo_wf_checkpoint_headers_on_workflow_id_thread_ts_id"})
-- remove_index(:p_duo_workflows_checkpoint_headers, {:name=>"index_duo_wf_checkpoint_headers_on_workflow_id"})
== 20260812160000 AddWorkflowIdThreadTsIndexToDuoWfCheckpointHeaders: migrated (0.2590s)

== 20260812160000 AddWorkflowIdThreadTsIndexToDuoWfCheckpointHeaders: reverting
-- add_index("gitlab_partitions_dynamic.p_duo_workflows_checkpoint_headers_20260713", :workflow_id, {:name=>"index_9be888bece", :algorithm=>:concurrently})
   (repeated per partition)
-- remove_index(:p_duo_workflows_checkpoint_headers, {:name=>"index_duo_wf_checkpoint_headers_on_workflow_id_thread_ts_id"})
== 20260812160000 AddWorkflowIdThreadTsIndexToDuoWfCheckpointHeaders: reverted

Rollback verified locally: down recreates the per-partition workflow_id indexes and drops the new one. db/structure.sql regenerated with scripts/regenerate-schema.

Pagination stays in the database (LIMIT/OFFSET). Blob reconstruction and the batched checkpoint_writes query run only for the rows on the returned page.

checkpoint_ns-filtered page and count

The index orders but cannot filter checkpoint_ns, so it lands in Filter rather than Index Cond. Plans below use one workflow with 60 headers alternating NULL / delegation:task-1, default page size.

Page (... AND checkpoint_ns IS NULL ORDER BY thread_ts DESC, id DESC LIMIT 20 OFFSET 0):

Limit  (cost=0.15..3.19 rows=1 width=220) (actual time=0.005..0.009 rows=20 loops=1)
  Buffers: shared hit=5
  ->  Index Scan Backward using index_duo_wf_checkpoint_headers_on_workflow_id_thread_ts_id on p_duo_workflows_checkpoint_headers_20260813
        Index Cond: (workflow_id = 1035)
        Filter: ((checkpoint_ns IS NULL) AND (workflow_created_at = '...'::timestamptz))
        Rows Removed by Filter: 20
Execution Time: 0.013 ms

Count:

Aggregate  (cost=3.19..3.20 rows=1 width=8) (actual time=0.012..0.012 rows=1 loops=1)
  Buffers: shared hit=2
  ->  Index Scan using index_duo_wf_checkpoint_headers_on_workflow_id_thread_ts_id on p_duo_workflows_checkpoint_headers_20260813
        Index Cond: (workflow_id = 1035)
        Filter: ((checkpoint_ns IS NULL) AND (workflow_created_at = '...'::timestamptz))
        Rows Removed by Filter: 30
Execution Time: 0.017 ms

The ordering is still index-served (no sort); the page walks the index backwards discarding non-matching rows, reading ~2x the page size at a 50/50 split. Both stay bounded by one workflow's headers within the single pruned partition, so checkpoint_ns is deliberately not in the index: it would only pay off when filtering out a large namespace, and the common path (the gateway omits checkpoint_ns) needs no such filter.