Add API endpoints and service for incremental checkpoint blobs

What does this MR do?

Adds the incremental Duo Workflow checkpoint blob read/write path on top of the range-partitioned p_duo_workflows_checkpoint_blobs table (introduced in the stacked DB MR !239245 (merged)):

• CreateCheckpointService writes per-channel blobs via bulk_insert!, base64-decoding the gateway payload into the bytea data column.
• Workflow#accumulated_blobs_for reads a workflow's blobs for the current current_thread group; the CheckpointBlob entity base64-encodes data and exposes the scalar id (the table has a composite [id, created_at] PK).
• Adds the current_thread column to duo_workflows_checkpoints and the incremental_checkpoints feature flag.

Stack: !239245 (merged) (DB) → this MR → !239249 (merged) (capability).

Database review

Both blob queries target the range-partitioned p_duo_workflows_checkpoint_blobs table.

Write — bulk_insert! (CreateCheckpointService#build_blobs_batch)

One INSERT per incremental checkpoint step; one row per changed channel (single-digit batch sizes):

INSERT INTO p_duo_workflows_checkpoint_blobs
  (workflow_id, project_id, namespace_id, current_thread, thread_ts, channel,
   version, write_type, step_action, data, created_at, updated_at)
VALUES (...), (...);

Insert on p_duo_workflows_checkpoint_blobs  (cost=0.00..0.03 rows=0 width=0)
  ->  Values Scan on "*VALUES*"  (cost=0.00..0.03 rows=2 width=244)

Rows are tuple-routed to the current day's partition by created_at; id is assigned by the table's BEFORE INSERT trigger.

Read — Workflow#accumulated_blobs_for

SELECT * FROM p_duo_workflows_checkpoint_blobs
WHERE workflow_id = $1 AND current_thread = $2
ORDER BY id;

EXPLAIN (ANALYZE, BUFFERS) on a local dataset of 10,000 rows / 500 workflows:

Sort  (cost=52.47..52.61 rows=58 width=225) (actual time=0.304..0.306 rows=7 loops=1)
  Sort Key: id
  Buffers: shared hit=62
  ->  Append  (cost=0.00..50.77 rows=58 width=225) (actual time=0.225..0.297 rows=7 loops=1)
        ->  Index Scan using p_duo_workflows_checkpoint_blobs_20260604_workflow_id_idx ...
              Index Cond: (workflow_id = 7)
              Filter: (current_thread = 0)
        ->  ...one index scan per partition...
 Execution Time: ~0.3 ms

Each partition is probed via index_duo_wf_checkpoint_blobs_on_workflow_id, with current_thread applied as a filter and the result sorted by id. The query is workflow-scoped, not time-scoped, so it reads across partitions by design (no created_at predicate / partition pruning). Per-workflow blob volume is small and bounded (steps × changed channels within one session), and rows are dropped after 30 days via partition drop, so the cross-partition Append stays cheap.

Plans generated locally via gdk psql; absolute costs are not production-representative.

Testing

1. Checkout Advertise incremental_checkpoints capability in... (!239249 - merged)
2. Enable the feature flag Feature.enable(:duo_workflow_incremental_checkpoints)
3. Trigger a flow (for example create merge request from an issue), and note it's id
4. The script below can be used to verify if data is stored properly

# bundle exec rails runner /tmp/verify_incremental_checkpoints.rb 582 

# frozen_string_literal: true
#
# Verify a Duo Workflow's incremental checkpoint blobs are stored properly:
#   1. integrity - counts, orphans, size limits, decode
#   2. storage   - non-incremental (full jsonb) vs incremental (blobs)
#
# Usage: bundle exec rails runner verify_incremental_checkpoints.rb <workflow_id>

require 'zlib'

WORKFLOW_ID = (ARGV[0] || 582).to_i
wf = Ai::DuoWorkflows::Workflow.find_by(id: WORKFLOW_ID)
abort "Workflow #{WORKFLOW_ID} not found" unless wf

def kb(b) = (b.to_f / 1024).round(1)

puts "=== Workflow #{wf.id} (status=#{wf.status_name}, project=#{wf.project_id}) ==="
puts "incremental_checkpoints_enabled?: #{wf.incremental_checkpoints_enabled?}"

# ---------------------------------------------------------------------------
# 1. INTEGRITY
# ---------------------------------------------------------------------------
cps   = wf.checkpoints
blobs = wf.checkpoint_blobs
puts "\n[1] INTEGRITY"
puts "  checkpoints: #{cps.count}   blobs: #{blobs.count}"
puts "  by write_type: #{blobs.group(:write_type).count}"
puts "  by step_action: #{blobs.group(:step_action).count}"
puts "  channels: #{blobs.distinct.pluck(:channel).sort.inspect}"

cp_ts   = cps.pluck(:thread_ts).to_set
blob_ts = blobs.distinct.pluck(:thread_ts).to_set
puts "  orphan blobs (no checkpoint): #{(blob_ts - cp_ts).to_a.inspect}"
puts "  checkpoints without blobs   : #{(cp_ts - blob_ts).size}"
limit = Ai::DuoWorkflows::CheckpointBlob::BLOB_DATA_LIMIT
puts "  blobs over #{kb(limit)} KB limit: #{blobs.where('octet_length(data) > ?', limit).count}"
puts "  null/empty data            : #{blobs.where('data IS NULL OR octet_length(data) = 0').count}"
# data is zlib-compressed msgpack; "stored properly" means it decompresses.
inflated_ok = blobs.find_each.count { |b| Zlib::Inflate.inflate(b.data) rescue false }
puts "  blobs that inflate cleanly : #{inflated_ok}/#{blobs.count}"

# ---------------------------------------------------------------------------
# 2. STORAGE
# ---------------------------------------------------------------------------
conn = ActiveRecord::Base.connection
cp_stored = conn.select_value(<<~SQL).to_i
  SELECT COALESCE(sum(pg_column_size(checkpoint) + pg_column_size(metadata)), 0)
  FROM p_duo_workflows_checkpoints WHERE workflow_id = #{wf.id}
SQL
bl_stored = conn.select_value(<<~SQL).to_i
  SELECT COALESCE(sum(pg_column_size(data)), 0)
  FROM p_duo_workflows_checkpoint_blobs WHERE workflow_id = #{wf.id}
SQL
puts "\n[2] STORAGE (on-disk, pg_column_size)"
puts "  non-incremental (full jsonb + metadata): #{kb cp_stored} KB"
puts "  incremental     (compressed blobs)     : #{kb bl_stored} KB"
puts "  ratio: #{(cp_stored.to_f / bl_stored).round(1)}x  (#{(100 * (1 - bl_stored.to_f / cp_stored)).round(1)}% smaller)" if bl_stored.positive?