Add ClickHouse finder for AI governance sessions

Hitesh Raghuvanshirequested to merge
630095-ai-governance-sessions-click-house-finder into master

What does this MR do and why?

Adds Ai::Governance::Sessions::ClickHouseFinder. It lists AI sessions, from GitLab Duo and from external agents such as Claude Code, for a group or project and everything under it, newest first. It reads from the ClickHouse table siphon_ai_governance_sessions.

GitLab.com will read AI governance sessions from ClickHouse first. The Postgres finder for self-managed is being built in parallel in !257556 (merged). Both finders take the same inputs and filters, so a later GraphQL field can pick either one.

  • Filters: agent class (GitLab Duo or external agents), flow type, user, project path, and a start time range. Flow type, user, and project path can each be included or excluded. Excluding a flow type keeps sessions that have no flow type.
  • Returns nothing unless the user has the read_agent_artifacts ability on the group or project, the same check as the Postgres finder.
  • Nothing calls this finder yet, so there is no user-facing change.
How the query reads the table
  • Siphon, the tool that copies Postgres rows into ClickHouse, can leave several versions of a row until ClickHouse merges them in the background. The finder picks the latest version of each row with argMax, grouped by the table's primary key (traversal_path, session_started_at, id), and drops rows marked as deleted.
  • Filters on primary key columns (namespace path, project path, start time) go in the inner query, so ClickHouse can skip data early.
  • Filters on other columns (source, flow type, user) apply after the latest version is picked, so an older version of a row can never match.
  • It follows the existing Ai::DuoWorkflows::SessionArtifacts::ClickHouseFinder. Like that finder, it returns a ClickHouse query builder rather than an ActiveRecord relation, so it is added to spec/support/finder_collection_allowlist.yml.

Database review

Nothing calls this finder yet. The plans below were taken on local ClickHouse (26.8) with synthetic data, following the ClickHouse reviewer guidelines:

  • 3M sessions, 4.8M row versions in siphon_ai_governance_sessions, spread over 365 days.
  • One large top-level group (1/9970/) with 300k sessions in 200 projects, each session with 2 versions. The other sessions are spread over 1,000 groups.
  • Each query adds LIMIT 20, as a GraphQL page would. Timings are from system.query_log on a warm run.
Case Elapsed Rows read (of 4.8M) Granules
A. Group, no filters 39 ms 600,736 295/2355
B. Group, last 30 days 35 ms 600,736 295/2355
C. Group and one project, last 30 days 13 ms 34,816 17/2355
D. Group, external agents, flow type excluded, one user 39 ms 600,736 295/2355
  • The group filter uses the primary key (traversal_path range), so only that group's rows are read.
  • session_started_at is the second primary key column. On its own it narrows reads only within one exact path, which is why B reads the same rows as A. With a project path it cuts reads to 17 granules (C).
  • Source, flow type, and user filters run after deduplication (D), so they don't change rows read.
A group, first page

params: {} | returned 20 rows

Elapsed: 39 ms | Read: 600736 of 4800000 rows (95.80 MiB) | Memory: 216.06 MiB

SELECT `siphon_ai_governance_sessions`.`id`, `siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`session_started_at`, `siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`traversal_path` FROM (SELECT `siphon_ai_governance_sessions`.`id`, argMax(`siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS namespace_id, argMax(`siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS project_id, argMax(`siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS user_id, argMax(`siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS workflow_id, argMax(`siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS created_at, argMax(`siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS updated_at, `siphon_ai_governance_sessions`.`session_started_at`, argMax(`siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS session_finished_at, argMax(`siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS source, argMax(`siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS status, argMax(`siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS external_xid, argMax(`siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS agent_type, argMax(`siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS flow_type, `siphon_ai_governance_sessions`.`traversal_path`, argMax(`siphon_ai_governance_sessions`.`_siphon_deleted`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS _siphon_deleted FROM `siphon_ai_governance_sessions` WHERE startsWith(`siphon_ai_governance_sessions`.`traversal_path`, '1/9970/') GROUP BY traversal_path, session_started_at, id) siphon_ai_governance_sessions WHERE `siphon_ai_governance_sessions`.`_siphon_deleted` = 'false' ORDER BY `siphon_ai_governance_sessions`.`session_started_at` DESC, `siphon_ai_governance_sessions`.`id` DESC LIMIT 20
Output: traversal_path, id, argMax(namespace_id, _siphon_replicated_at), _siphon_replicated_at, argMax(project_id, _siphon_replicated_at), argMax(user_id, _siphon_replicated_at), argMax(workflow_id, _siphon_replicated_at), argMax(created_at, _siphon_replicated_at), argMax(updated_at, _siphon_replicated_at), session_started_at, argMax(session_finished_at, _siphon_replicated_at), argMax(source, _siphon_replicated_at), argMax(status, _siphon_replicated_at), argMax(external_xid, _siphon_replicated_at), argMax(agent_type, _siphon_replicated_at), argMax(flow_type, _siphon_replicated_at), _siphon_deleted
Prewhere filter
Prewhere filter column:  startsWith(traversal_path, '1/9970/')
Indexes:
PrimaryKey
Keys:
traversal_path
Condition: (traversal_path in ['1/9970/', '1/99700'))
Parts: 3/5
Granules: 295/2355
B group, last 30 days

params: {"session_started_after":"2026-08-29T17:37:52.161Z","session_started_before":"2026-09-28T17:37:52.161Z"} | returned 20 rows

Elapsed: 35 ms | Read: 600736 of 4800000 rows (95.56 MiB) | Memory: 181.63 MiB

SELECT `siphon_ai_governance_sessions`.`id`, `siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`session_started_at`, `siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`traversal_path` FROM (SELECT `siphon_ai_governance_sessions`.`id`, argMax(`siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS namespace_id, argMax(`siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS project_id, argMax(`siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS user_id, argMax(`siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS workflow_id, argMax(`siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS created_at, argMax(`siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS updated_at, `siphon_ai_governance_sessions`.`session_started_at`, argMax(`siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS session_finished_at, argMax(`siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS source, argMax(`siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS status, argMax(`siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS external_xid, argMax(`siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS agent_type, argMax(`siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS flow_type, `siphon_ai_governance_sessions`.`traversal_path`, argMax(`siphon_ai_governance_sessions`.`_siphon_deleted`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS _siphon_deleted FROM `siphon_ai_governance_sessions` WHERE startsWith(`siphon_ai_governance_sessions`.`traversal_path`, '1/9970/') AND `siphon_ai_governance_sessions`.`session_started_at` >= '2026-08-29 17:37:52.161515' AND `siphon_ai_governance_sessions`.`session_started_at` <= '2026-09-28 17:37:52.161515' GROUP BY traversal_path, session_started_at, id) siphon_ai_governance_sessions WHERE `siphon_ai_governance_sessions`.`_siphon_deleted` = 'false' ORDER BY `siphon_ai_governance_sessions`.`session_started_at` DESC, `siphon_ai_governance_sessions`.`id` DESC LIMIT 20
Output: session_started_at, traversal_path, id, argMax(namespace_id, _siphon_replicated_at), _siphon_replicated_at, argMax(project_id, _siphon_replicated_at), argMax(user_id, _siphon_replicated_at), argMax(workflow_id, _siphon_replicated_at), argMax(created_at, _siphon_replicated_at), argMax(updated_at, _siphon_replicated_at), argMax(session_finished_at, _siphon_replicated_at), argMax(source, _siphon_replicated_at), argMax(status, _siphon_replicated_at), argMax(external_xid, _siphon_replicated_at), argMax(agent_type, _siphon_replicated_at), argMax(flow_type, _siphon_replicated_at), _siphon_deleted
Prewhere filter
Prewhere filter column:  startsWith(traversal_path, '1/9970/') AND session_started_at <= '2026-09-28 17:37:52.161515' AND session_started_at >= '2026-08-29 17:37:52.161515'
Indexes:
PrimaryKey
Keys:
traversal_path
session_started_at
Condition: and((session_started_at in (-Inf, '1790617072.161515']), and((session_started_at in ['1788025072.161515', +Inf)), (traversal_path in ['1/9970/', '1/99700'))))
Parts: 3/5
Granules: 295/2355
C group + project, last 30 days

params: {"project_path":"gitlab-duo/secret_fp_5","session_started_after":"2026-08-29T17:37:52.161Z","session_started_before":"2026-09-28T17:37:52.161Z"} | returned 20 rows

Elapsed: 13 ms | Read: 34816 of 4800000 rows (5.05 MiB) | Memory: 33.74 MiB

SELECT `siphon_ai_governance_sessions`.`id`, `siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`session_started_at`, `siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`traversal_path` FROM (SELECT `siphon_ai_governance_sessions`.`id`, argMax(`siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS namespace_id, argMax(`siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS project_id, argMax(`siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS user_id, argMax(`siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS workflow_id, argMax(`siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS created_at, argMax(`siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS updated_at, `siphon_ai_governance_sessions`.`session_started_at`, argMax(`siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS session_finished_at, argMax(`siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS source, argMax(`siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS status, argMax(`siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS external_xid, argMax(`siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS agent_type, argMax(`siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS flow_type, `siphon_ai_governance_sessions`.`traversal_path`, argMax(`siphon_ai_governance_sessions`.`_siphon_deleted`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS _siphon_deleted FROM `siphon_ai_governance_sessions` WHERE startsWith(`siphon_ai_governance_sessions`.`traversal_path`, '1/9970/') AND startsWith(`siphon_ai_governance_sessions`.`traversal_path`, '1/9970/100050/') AND `siphon_ai_governance_sessions`.`session_started_at` >= '2026-08-29 17:37:52.161515' AND `siphon_ai_governance_sessions`.`session_started_at` <= '2026-09-28 17:37:52.161515' GROUP BY traversal_path, session_started_at, id) siphon_ai_governance_sessions WHERE `siphon_ai_governance_sessions`.`_siphon_deleted` = 'false' ORDER BY `siphon_ai_governance_sessions`.`session_started_at` DESC, `siphon_ai_governance_sessions`.`id` DESC LIMIT 20
Output: session_started_at, traversal_path, id, argMax(namespace_id, _siphon_replicated_at), _siphon_replicated_at, argMax(project_id, _siphon_replicated_at), argMax(user_id, _siphon_replicated_at), argMax(workflow_id, _siphon_replicated_at), argMax(created_at, _siphon_replicated_at), argMax(updated_at, _siphon_replicated_at), argMax(session_finished_at, _siphon_replicated_at), argMax(source, _siphon_replicated_at), argMax(status, _siphon_replicated_at), argMax(external_xid, _siphon_replicated_at), argMax(agent_type, _siphon_replicated_at), argMax(flow_type, _siphon_replicated_at), _siphon_deleted
Prewhere filter
Prewhere filter column:  startsWith(traversal_path, '1/9970/100050/') AND startsWith(traversal_path, '1/9970/') AND session_started_at <= '2026-09-28 17:37:52.161515' AND session_started_at >= '2026-08-29 17:37:52.161515'
Indexes:
PrimaryKey
Keys:
traversal_path
session_started_at
Condition: and((session_started_at in (-Inf, '1790617072.161515']), and((session_started_at in ['1788025072.161515', +Inf)), and((traversal_path in ['1/9970/100050/', '1/9970/1000500')), (traversal_path in ['1/9970/', '1/99700')))))
Parts: 3/5
Granules: 17/2355
D group, external + NOT flow + user

params: {"agent_class":"external","not":{"flow_type":"x"},"user_id":30} | returned 20 rows

Elapsed: 39 ms | Read: 600736 of 4800000 rows (95.80 MiB) | Memory: 216.76 MiB

SELECT `siphon_ai_governance_sessions`.`id`, `siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`session_started_at`, `siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`traversal_path` FROM (SELECT `siphon_ai_governance_sessions`.`id`, argMax(`siphon_ai_governance_sessions`.`namespace_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS namespace_id, argMax(`siphon_ai_governance_sessions`.`project_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS project_id, argMax(`siphon_ai_governance_sessions`.`user_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS user_id, argMax(`siphon_ai_governance_sessions`.`workflow_id`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS workflow_id, argMax(`siphon_ai_governance_sessions`.`created_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS created_at, argMax(`siphon_ai_governance_sessions`.`updated_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS updated_at, `siphon_ai_governance_sessions`.`session_started_at`, argMax(`siphon_ai_governance_sessions`.`session_finished_at`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS session_finished_at, argMax(`siphon_ai_governance_sessions`.`source`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS source, argMax(`siphon_ai_governance_sessions`.`status`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS status, argMax(`siphon_ai_governance_sessions`.`external_xid`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS external_xid, argMax(`siphon_ai_governance_sessions`.`agent_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS agent_type, argMax(`siphon_ai_governance_sessions`.`flow_type`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS flow_type, `siphon_ai_governance_sessions`.`traversal_path`, argMax(`siphon_ai_governance_sessions`.`_siphon_deleted`, `siphon_ai_governance_sessions`.`_siphon_replicated_at`) AS _siphon_deleted FROM `siphon_ai_governance_sessions` WHERE startsWith(`siphon_ai_governance_sessions`.`traversal_path`, '1/9970/') GROUP BY traversal_path, session_started_at, id) siphon_ai_governance_sessions WHERE `siphon_ai_governance_sessions`.`_siphon_deleted` = 'false' AND `siphon_ai_governance_sessions`.`source` IN (1, 2) AND (`siphon_ai_governance_sessions`.`flow_type` IS NULL OR `siphon_ai_governance_sessions`.`flow_type` != 'x') AND `siphon_ai_governance_sessions`.`user_id` = 30 ORDER BY `siphon_ai_governance_sessions`.`session_started_at` DESC, `siphon_ai_governance_sessions`.`id` DESC LIMIT 20
Output: traversal_path, id, argMax(namespace_id, _siphon_replicated_at), _siphon_replicated_at, argMax(project_id, _siphon_replicated_at), argMax(user_id, _siphon_replicated_at), argMax(workflow_id, _siphon_replicated_at), argMax(created_at, _siphon_replicated_at), argMax(updated_at, _siphon_replicated_at), session_started_at, argMax(session_finished_at, _siphon_replicated_at), argMax(source, _siphon_replicated_at), argMax(status, _siphon_replicated_at), argMax(external_xid, _siphon_replicated_at), argMax(agent_type, _siphon_replicated_at), argMax(flow_type, _siphon_replicated_at), _siphon_deleted
Prewhere filter
Prewhere filter column:  startsWith(traversal_path, '1/9970/')
Indexes:
PrimaryKey
Keys:
traversal_path
Condition: (traversal_path in ['1/9970/', '1/99700'))
Parts: 3/5
Granules: 295/2355

References

Screenshots or screen recordings

Not applicable. Backend only.

How to set up and validate locally

With ClickHouse enabled in GDK, run:

bundle exec rspec ee/spec/finders/ai/governance/sessions/click_house_finder_spec.rb

The spec inserts real rows into ClickHouse, including an older and a newer version of the same row and a deleted row. So deduplication, filters, and ordering are tested against ClickHouse itself, not only the SQL text. Locally: 19 examples, 0 failures.

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 Hitesh Raghuvanshi

Merge request reports