From 2a99970ff86b96493d0e22ed585db7abd41cfef6 Mon Sep 17 00:00:00 2001 From: aadereiko Date: Wed, 19 Aug 2026 14:39:09 +0200 Subject: [PATCH 1/4] feat(cipx-spends): persist trigger, trigger_detail, turn_key and parent_tool_use_id cipx already ships all four on metadata.cipx.call, but none reached a typed column, so the spend tables cannot tell a subagent's spend from the main agent's, name the agent, or reconstruct the parent/child tree. Migration 000118 adds them in one ALTER, additive with DEFAULT '' so pre-existing rows read as unknown - honest, since they were written before the proxy carried the fields. trigger_detail is the agent NAME when trigger='subagent' and the tool name when trigger='tool_continuation', so consumers must gate on trigger before grouping per agent. Empty is deliberately not defaulted: it means unknown, never a guessed agent name. Co-Authored-By: Claude Opus 5 (1M context) --- .../com/comet/opik/domain/CipxSpendDAO.java | 40 +++++++++++--- .../000118_add_trigger_to_cipx_spends.sql | 30 +++++++++++ .../events/CostIntelligenceIngestionTest.java | 52 +++++++++++++++++-- 3 files changed, 111 insertions(+), 11 deletions(-) create mode 100644 apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/CipxSpendDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/CipxSpendDAO.java index 90e1a2389c7..db79203f3c2 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/CipxSpendDAO.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/CipxSpendDAO.java @@ -25,8 +25,9 @@ import static com.comet.opik.utils.template.TemplateUtils.getQueryItemPlaceHolder; /** - * Writes the cipx_spends table from cipx LLM-call spans: span-level call data only (model + usage - * counters); the blocks land in cipx_spend_blocks via {@link CipxSpendBlockDAO}. Triggered + * Writes the cipx_spends table from cipx LLM-call spans: span-level call data only (model, usage + * counters, config knobs and the call's trigger/attribution fields); the blocks land in + * cipx_spend_blocks via {@link CipxSpendBlockDAO}. Triggered * asynchronously off span create events; never reads the spans or cipx_spends tables. The cipx fields * are parsed from metadata in Java ({@link SpanRow#from}); the listener only passes rows it has * already gated to cipx. @@ -60,7 +61,11 @@ public record SpanRow( @NonNull String thinkingType, long maxTokens, @NonNull String contextManagement, - @NonNull String speed) { + @NonNull String speed, + @NonNull String trigger, + @NonNull String triggerDetail, + @NonNull String turnKey, + @NonNull String parentToolUseId) { public static SpanRow from(UUID spanId, UUID traceId, UUID projectId, JsonNode metadata, Instant startTime) { JsonNode call = metadata.path("cipx").path("call"); @@ -84,6 +89,16 @@ public static SpanRow from(UUID spanId, UUID traceId, UUID projectId, JsonNode m .maxTokens(config.path("max_tokens").asLong(0)) .contextManagement(config.path("context_management").asText("")) .speed(config.path("speed").asText("")) + // Attribution fields. trigger_detail carries the subagent NAME when + // trigger=subagent (the parent's Agent tool_use input.subagent_type) and the tool + // name when trigger=tool_continuation; parent_tool_use_id identifies the agent + // invocation this call belongs to, which is the parent/child edge of the agent + // tree. All default to "" when the proxy could not resolve them — an empty value + // means unknown and must not be substituted for a guessed agent name. + .trigger(call.path("trigger").asText("")) + .triggerDetail(call.path("trigger_detail").asText("")) + .turnKey(call.path("turn_key").asText("")) + .parentToolUseId(call.path("parent_tool_use_id").asText("")) .build(); } } @@ -94,7 +109,8 @@ public static SpanRow from(UUID spanId, UUID traceId, UUID projectId, JsonNode m INSERT INTO cipx_spends (workspace_id, project_id, trace_id, span_id, start_time, model, u_input, u_cache_read, u_cache_creation, u_cache_creation_5m, u_cache_creation_1h, u_output, - effort, thinking_type, max_tokens, context_management, speed) + effort, thinking_type, max_tokens, context_management, speed, + `trigger`, trigger_detail, turn_key, parent_tool_use_id) SETTINGS log_comment = '' FORMAT Values , :max_tokens, :context_management, - :speed + :speed, + :trigger, + :trigger_detail, + :turn_key, + :parent_tool_use_id ) , }> @@ -144,7 +164,9 @@ private Publisher insert(List rows, String workspaceI // Positional binds: the driver resolves named binds with a linear indexOf over the statement's // parameter list (quadratic per statement), while bind(int) is a direct array write. Indices // follow the placeholders' first-appearance order in the rendered SQL: workspace_id once at 0 - // (repeats dedup), then 16 parameters per row tuple in template order. + // (repeats dedup), then 20 parameters per row tuple in template order. The bind order below + // must stay in lockstep with the INSERT tuple above — nothing checks it at compile time, and a + // mismatch silently writes each value into the neighbouring column. statement.bind(0, workspaceId); int index = 1; for (SpanRow row : rows) { @@ -163,7 +185,11 @@ private Publisher insert(List rows, String workspaceI .bind(index++, row.thinkingType()) .bind(index++, row.maxTokens()) .bind(index++, row.contextManagement()) - .bind(index++, row.speed()); + .bind(index++, row.speed()) + .bind(index++, row.trigger()) + .bind(index++, row.triggerDetail()) + .bind(index++, row.turnKey()) + .bind(index++, row.parentToolUseId()); } return statement.execute(); diff --git a/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql b/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql new file mode 100644 index 00000000000..18afaaf2c74 --- /dev/null +++ b/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql @@ -0,0 +1,30 @@ +--liquibase formatted sql +--changeset aadereiko:000118_add_trigger_to_cipx_spends +--comment: Persist the per-call trigger, its detail (subagent NAME), the turn key and the parent tool_use id on cipx_spends +-- +-- cipx already ships all four on metadata.cipx.call; none of them reached a typed column, so the +-- spend tables cannot tell a subagent's spend from the main agent's, let alone name the agent or +-- reconstruct the parent/child tree the AI Spend UI wants. +-- +-- trigger what caused this call: user_turn | tool_continuation | subagent | automated +-- | unknown. Closed enum -> LowCardinality. +-- trigger_detail trigger-dependent qualifier. For trigger='subagent' this is the agent NAME +-- (the parent's `Agent` tool_use input.subagent_type: "code-reviewer", +-- "Explore", ...); for trigger='tool_continuation' it is the tool name. Empty +-- when the proxy could not resolve it -- deliberately NOT defaulted, so an +-- empty string means "unknown", never "general-purpose". +-- turn_key groups a user prompt's root call with every continuation that followed it +-- (SHA256 hex of the prompt text). The grain a per-turn read aggregates on. +-- parent_tool_use_id the `Agent` tool_use that spawned this call -- the identity of one agent +-- INVOCATION, stable across every call that agent makes. This is the edge of +-- the agent tree: child rows point at the parent's tool_use id. +-- +-- Additive columns with defaults; every pre-existing row reads '' (unknown), which is honest -- +-- those rows were written before the proxy carried the fields. +ALTER TABLE ${ANALYTICS_DB_DATABASE_NAME}.cipx_spends ON CLUSTER '{cluster}' + ADD COLUMN IF NOT EXISTS `trigger` LowCardinality(String) DEFAULT '', + ADD COLUMN IF NOT EXISTS trigger_detail LowCardinality(String) DEFAULT '', + ADD COLUMN IF NOT EXISTS turn_key String DEFAULT '', + ADD COLUMN IF NOT EXISTS parent_tool_use_id String DEFAULT ''; + +--rollback ALTER TABLE ${ANALYTICS_DB_DATABASE_NAME}.cipx_spends ON CLUSTER '{cluster}' DROP COLUMN IF EXISTS `trigger`, DROP COLUMN IF EXISTS trigger_detail, DROP COLUMN IF EXISTS turn_key, DROP COLUMN IF EXISTS parent_tool_use_id; diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java index bf734299fa7..600f1d384f3 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java @@ -149,6 +149,13 @@ void spanCreatedWithCipxCallLands() { assertThat(row.get().contextManagement()).isEqualTo("clear_thinking_20251015"); // speed: selects the rate table, so it must survive ingestion assertThat(row.get().speed()).isEqualTo("fast"); + // attribution fields: what caused the call, which agent ran it, which turn it + // belongs to, and which Agent tool_use spawned it. Without these the spend tables + // cannot separate subagent spend from the main agent's, nor name the agent. + assertThat(row.get().trigger()).isEqualTo("subagent"); + assertThat(row.get().triggerDetail()).isEqualTo("code-reviewer"); + assertThat(row.get().turnKey()).isEqualTo("abc123turnkey"); + assertThat(row.get().parentToolUseId()).isEqualTo("toolu_parent_agent"); }); // Carried on every block row too. @@ -162,6 +169,33 @@ void spanCreatedWithCipxCallLands() { assertThat(getCipxBlocks(plainSpan.id(), ws.workspaceId())).isEmpty(); } + @Test + @DisplayName("a call carrying no trigger fields ingests with empty attribution columns") + void spanWithoutTriggerFieldsLandsWithEmptyAttribution() { + var ws = newWorkspace(); + String projectName = "cipx-" + UUID.randomUUID(); + + // systemToolsCipxMetadata carries no trigger/turn_key/parent_tool_use_id — the shape + // every span written before the proxy shipped them has. Ingestion must still land the + // row and must leave the attribution columns empty rather than substituting a default: + // "" means unknown, and a guessed agent name would book real spend against an agent + // that never ran. + var span = factory.manufacturePojo(Span.class).toBuilder() + .projectName(projectName) + .metadata(systemToolsCipxMetadata("claude-sonnet-4-6", 200)) + .build(); + spanResourceClient.createSpan(span, ws.apiKey(), ws.workspaceName()); + + await().atMost(30, SECONDS).untilAsserted(() -> { + var row = getCipxSpend(span.id(), ws.workspaceId()); + assertThat(row).isPresent(); + assertThat(row.get().trigger()).isEmpty(); + assertThat(row.get().triggerDetail()).isEmpty(); + assertThat(row.get().turnKey()).isEmpty(); + assertThat(row.get().parentToolUseId()).isEmpty(); + }); + } + @Test @DisplayName("blocks land with derived allocation, residual rows, and identity_context dropped") void blocksLandWithDerivedAllocationAndResiduals() { @@ -542,7 +576,11 @@ private static JsonNode spanCipxMetadata(String model, long input, long cacheRea "max_tokens": 64000, "context_management": "clear_thinking_20251015", "speed": "fast" - } + }, + "trigger": "subagent", + "trigger_detail": "code-reviewer", + "turn_key": "abc123turnkey", + "parent_tool_use_id": "toolu_parent_agent" }, "blocks": [ {"category":"memory","side":"input","cache_status":"read","parent_category":"context","chars":120,"tool_name":"","tool_server":"","tool_use_id":"","resource":"CLAUDE.md","kind":"text","subcategory":"auto_memory","sha256":"a1b2c3"}, @@ -666,7 +704,8 @@ private Optional getCipxSpend(UUID spanId, String workspaceId) { toUnixTimestamp64Milli(start_time) AS start_ms, model AS model, u_input, u_cache_read, u_cache_creation, u_cache_creation_5m, u_cache_creation_1h, u_output, - effort, thinking_type, max_tokens, context_management, speed + effort, thinking_type, max_tokens, context_management, speed, + `trigger` AS trigger_kind, trigger_detail, turn_key, parent_tool_use_id FROM cipx_spends FINAL WHERE workspace_id = :workspace_id AND span_id = :span_id """; @@ -689,7 +728,11 @@ private Optional getCipxSpend(UUID spanId, String workspaceId) { row.get("thinking_type", String.class), row.get("max_tokens", Long.class), row.get("context_management", String.class), - row.get("speed", String.class))))); + row.get("speed", String.class), + row.get("trigger_kind", String.class), + row.get("trigger_detail", String.class), + row.get("turn_key", String.class), + row.get("parent_tool_use_id", String.class))))); }).blockOptional(); } @@ -814,7 +857,8 @@ private record WorkspaceContext(String apiKey, String workspaceName, String work private record CipxSpendRow(String projectId, Long startMs, String model, Long uInput, Long uCacheRead, Long uCacheCreation, Long uCacheCreation5m, Long uCacheCreation1h, Long uOutput, String effort, - String thinkingType, Long maxTokens, String contextManagement, String speed) { + String thinkingType, Long maxTokens, String contextManagement, String speed, String trigger, + String triggerDetail, String turnKey, String parentToolUseId) { } private record CipxBlockRow(Integer blockIdx, String src, String category, String tier, String lane, From ade7d8ab822479b1e19e33247fdcf040b6de51b5 Mon Sep 17 00:00:00 2001 From: aadereiko Date: Thu, 20 Aug 2026 17:56:12 +0200 Subject: [PATCH 2/4] test(cipx-spends): pin the positional bind order and unknown-field tolerance CipxSpendDAO binds by position and nothing checks the order against the INSERT tuple at compile time - the DAO's own comment says a mismatch "silently writes each value into the neighbouring column". Give every column a distinct sentinel and assert each holds its own, with the column name as the assertion description so a regression names the column that received the wrong value. Two rows, not one: the bind index accumulates across rows while workspace_id is bound once at index 0 and its repeats dedup. If that dedup assumption is ever wrong the stride becomes 21 instead of 20 and only the SECOND row is corrupted, which a single-row insert cannot see. The ids are UUIDs because project_id/trace_id/span_id are FixedString(36), so a rotation among those three is silently accepted by ClickHouse. Second test: cipx adds fields to metadata.cipx.call on its own release cadence and ships to laptops independently of this service, so a newer proxy talking to an older backend is the normal state. Unknown fields at every level - on the call, inside usage (inference_geo, the real pending one from OPIK-7757), inside cache_creation, inside config, beside call, on a block, and beside cipx in metadata - must be ignored rather than reject the row. Dropping the row would lose the spend entirely, and spend totals are the one thing already correct. Co-Authored-By: Claude Opus 5 (1M context) --- .../events/CostIntelligenceIngestionTest.java | 236 +++++++++++++++++- 1 file changed, 235 insertions(+), 1 deletion(-) diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java index 600f1d384f3..fd1afcf2c46 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java @@ -14,6 +14,7 @@ import com.comet.opik.api.resources.utils.WireMockUtils; import com.comet.opik.api.resources.utils.resources.SpanResourceClient; import com.comet.opik.api.resources.utils.resources.TraceResourceClient; +import com.comet.opik.domain.CipxSpendDAO; import com.comet.opik.extensions.DropwizardAppExtensionProvider; import com.comet.opik.extensions.RegisterApp; import com.comet.opik.infrastructure.db.TransactionTemplateAsync; @@ -42,6 +43,7 @@ import ru.vyarus.guicey.jdbi3.tx.TransactionTemplate; import uk.co.jemos.podam.api.PodamFactory; +import java.time.Instant; import java.util.List; import java.util.Optional; import java.util.UUID; @@ -94,13 +96,15 @@ class CostIntelligenceIngestionTest { private TransactionTemplate mySqlTemplate; private SpanResourceClient spanResourceClient; private TraceResourceClient traceResourceClient; + private CipxSpendDAO cipxSpendDAO; @BeforeAll void setUpAll(ClientSupport client, TransactionTemplateAsync clickHouseTemplate, - TransactionTemplate mySqlTemplate) { + TransactionTemplate mySqlTemplate, CipxSpendDAO cipxSpendDAO) { this.baseURI = TestUtils.getBaseUrl(client); this.clickHouseTemplate = clickHouseTemplate; this.mySqlTemplate = mySqlTemplate; + this.cipxSpendDAO = cipxSpendDAO; ClientSupportUtils.config(client); @@ -196,6 +200,71 @@ void spanWithoutTriggerFieldsLandsWithEmptyAttribution() { }); } + @Test + @DisplayName("every positional bind lands in its own column, across a multi-row insert") + void everyPositionalBindLandsInItsOwnColumn() { + var ws = newWorkspace(); + + // CipxSpendDAO binds by position and nothing checks the bind order against the INSERT + // tuple at compile time. A mismatch does not fail the insert — it writes each value into + // the neighbouring column, which is invisible unless the columns hold values that can be + // told apart. So give every column its own recognizable sentinel and assert each one + // holds its own: a rotation then surfaces as a value reported under the wrong name. + // + // Two rows, deliberately: the bind index accumulates across rows while workspace_id is + // bound once at index 0 and its repeats dedup. If that dedup assumption is ever wrong + // the stride becomes 21 instead of 20 and only the second row is corrupted — which a + // single-row insert cannot see. + var rowOne = sentinelRow(1); + var rowTwo = sentinelRow(2); + + cipxSpendDAO.insert(List.of(rowOne, rowTwo), ws.workspaceId(), USER).block(); + + assertEveryColumnHoldsItsOwnValue(ws.workspaceId(), rowOne); + assertEveryColumnHoldsItsOwnValue(ws.workspaceId(), rowTwo); + } + + @Test + @DisplayName("a call carrying fields the DAO does not know still ingests") + void unknownFieldsDoNotRejectTheRow() { + var ws = newWorkspace(); + String projectName = "cipx-" + UUID.randomUUID(); + + // cipx adds fields to metadata.cipx.call on its own release cadence and ships to laptops + // independently of this service, so a newer proxy talking to an older backend is the + // normal state, not an edge case. Ingestion must ignore what it does not recognize + // rather than reject the row — dropping the row would lose the spend entirely, and spend + // totals are the one thing that is correct today. + var span = factory.manufacturePojo(Span.class).toBuilder() + .projectName(projectName) + .metadata(unknownFieldsCipxMetadata("claude-sonnet-4-6")) + .build(); + spanResourceClient.createSpan(span, ws.apiKey(), ws.workspaceName()); + + await().atMost(30, SECONDS).untilAsserted(() -> { + var row = getCipxSpend(span.id(), ws.workspaceId()); + assertThat(row).isPresent(); + // Every known field still parsed correctly alongside the unknown ones. + assertThat(row.get().model()).isEqualTo("claude-sonnet-4-6"); + assertThat(row.get().uInput()).isEqualTo(11L); + assertThat(row.get().uCacheRead()).isEqualTo(22L); + assertThat(row.get().uCacheCreation()).isEqualTo(33L); + assertThat(row.get().uCacheCreation5m()).isEqualTo(44L); + assertThat(row.get().uCacheCreation1h()).isEqualTo(55L); + assertThat(row.get().uOutput()).isEqualTo(66L); + assertThat(row.get().effort()).isEqualTo("high"); + assertThat(row.get().speed()).isEqualTo("fast"); + assertThat(row.get().trigger()).isEqualTo("subagent"); + assertThat(row.get().triggerDetail()).isEqualTo("Explore"); + assertThat(row.get().turnKey()).isEqualTo("unknown-fields-turnkey"); + assertThat(row.get().parentToolUseId()).isEqualTo("toolu_unknown_fields"); + }); + + // The block writer sees the same metadata, so an unknown field on a block must not drop + // the blocks either. + assertThat(getCipxBlocks(span.id(), ws.workspaceId())).isNotEmpty(); + } + @Test @DisplayName("blocks land with derived allocation, residual rows, and identity_context dropped") void blocksLandWithDerivedAllocationAndResiduals() { @@ -552,6 +621,164 @@ private WorkspaceContext newWorkspace() { return new WorkspaceContext(apiKey, workspaceName, workspaceId); } + // One distinct value per column, derived from n so two rows never collide. The ids are UUIDs + // because project_id/trace_id/span_id are FixedString(36) — a rotation among those three is + // silently accepted by ClickHouse, which is precisely why they need telling apart. + private CipxSpendDAO.SpanRow sentinelRow(int n) { + long base = n * 1_000_000L; + return CipxSpendDAO.SpanRow.builder() + .projectId(UUID.randomUUID().toString()) + .traceId(UUID.randomUUID().toString()) + .spanId(UUID.randomUUID().toString()) + .startTime(Instant.ofEpochMilli(1_800_000_000_000L + n)) + .model("sentinel-" + n + "-model") + .uInput(base + 1) + .uCacheRead(base + 2) + .uCacheCreation(base + 3) + .uCacheCreation5m(base + 4) + .uCacheCreation1h(base + 5) + .uOutput(base + 6) + .effort("sentinel-" + n + "-effort") + .thinkingType("sentinel-" + n + "-thinking-type") + .maxTokens(base + 7) + .contextManagement("sentinel-" + n + "-context-management") + .speed("sentinel-" + n + "-speed") + .trigger("sentinel-" + n + "-trigger") + .triggerDetail("sentinel-" + n + "-trigger-detail") + .turnKey("sentinel-" + n + "-turn-key") + .parentToolUseId("sentinel-" + n + "-parent-tool-use-id") + .build(); + } + + // Asserts column by column with the column name as the description, so a bind-order regression + // reports which column received the wrong value rather than just "expected X but was Y". + private void assertEveryColumnHoldsItsOwnValue(String workspaceId, CipxSpendDAO.SpanRow expected) { + var stored = getCipxSpendAllColumns(expected.spanId(), workspaceId); + assertThat(stored).as("row for span_id %s", expected.spanId()).isPresent(); + var actual = stored.get(); + + assertThat(actual.workspaceId()).as("workspace_id").isEqualTo(workspaceId); + assertThat(actual.projectId()).as("project_id").isEqualTo(expected.projectId()); + assertThat(actual.traceId()).as("trace_id").isEqualTo(expected.traceId()); + assertThat(actual.spanId()).as("span_id").isEqualTo(expected.spanId()); + assertThat(actual.startMs()).as("start_time").isEqualTo(expected.startTime().toEpochMilli()); + assertThat(actual.model()).as("model").isEqualTo(expected.model()); + assertThat(actual.uInput()).as("u_input").isEqualTo(expected.uInput()); + assertThat(actual.uCacheRead()).as("u_cache_read").isEqualTo(expected.uCacheRead()); + assertThat(actual.uCacheCreation()).as("u_cache_creation").isEqualTo(expected.uCacheCreation()); + assertThat(actual.uCacheCreation5m()).as("u_cache_creation_5m").isEqualTo(expected.uCacheCreation5m()); + assertThat(actual.uCacheCreation1h()).as("u_cache_creation_1h").isEqualTo(expected.uCacheCreation1h()); + assertThat(actual.uOutput()).as("u_output").isEqualTo(expected.uOutput()); + assertThat(actual.effort()).as("effort").isEqualTo(expected.effort()); + assertThat(actual.thinkingType()).as("thinking_type").isEqualTo(expected.thinkingType()); + assertThat(actual.maxTokens()).as("max_tokens").isEqualTo(expected.maxTokens()); + assertThat(actual.contextManagement()).as("context_management").isEqualTo(expected.contextManagement()); + assertThat(actual.speed()).as("speed").isEqualTo(expected.speed()); + assertThat(actual.trigger()).as("trigger").isEqualTo(expected.trigger()); + assertThat(actual.triggerDetail()).as("trigger_detail").isEqualTo(expected.triggerDetail()); + assertThat(actual.turnKey()).as("turn_key").isEqualTo(expected.turnKey()); + assertThat(actual.parentToolUseId()).as("parent_tool_use_id").isEqualTo(expected.parentToolUseId()); + } + + // Reads every column the DAO writes, including workspace_id and trace_id which the narrower + // getCipxSpend does not project. Bind-order coverage is only as wide as the read. + private Optional getCipxSpendAllColumns(String spanId, String workspaceId) { + String sql = """ + SELECT + workspace_id AS workspace_id, + project_id AS project_id, + trace_id AS trace_id, + span_id AS span_id, + toUnixTimestamp64Milli(start_time) AS start_ms, + model AS model, + u_input, u_cache_read, u_cache_creation, u_cache_creation_5m, u_cache_creation_1h, u_output, + effort, thinking_type, max_tokens, context_management, speed, + `trigger` AS trigger_kind, trigger_detail, turn_key, parent_tool_use_id + FROM cipx_spends FINAL + WHERE workspace_id = :workspace_id AND span_id = :span_id + """; + return clickHouseTemplate.nonTransaction(connection -> { + var statement = connection.createStatement(sql) + .bind("workspace_id", workspaceId) + .bind("span_id", spanId); + return Mono.from(statement.execute()) + .flatMap(result -> Mono.from(result.map((row, meta) -> new SentinelSpendRow( + row.get("workspace_id", String.class), + row.get("project_id", String.class), + row.get("trace_id", String.class), + row.get("span_id", String.class), + row.get("start_ms", Long.class), + row.get("model", String.class), + row.get("u_input", Long.class), + row.get("u_cache_read", Long.class), + row.get("u_cache_creation", Long.class), + row.get("u_cache_creation_5m", Long.class), + row.get("u_cache_creation_1h", Long.class), + row.get("u_output", Long.class), + row.get("effort", String.class), + row.get("thinking_type", String.class), + row.get("max_tokens", Long.class), + row.get("context_management", String.class), + row.get("speed", String.class), + row.get("trigger_kind", String.class), + row.get("trigger_detail", String.class), + row.get("turn_key", String.class), + row.get("parent_tool_use_id", String.class))))); + }).blockOptional(); + } + + // A cipx call carrying fields this backend has never heard of, at every level a newer proxy could + // add them: on the call, inside usage (inference_geo is the real pending one — OPIK-7757), inside + // cache_creation, inside config, as a sibling of call under cipx, on a block, and beside cipx in + // metadata. Every known field is still present and must still parse. + private static JsonNode unknownFieldsCipxMetadata(String model) { + return JsonUtils.getJsonNodeFromString( + """ + { + "unrelated_top_level": {"anything": true}, + "cipx": { + "future_section": {"whatever": [1, 2, 3]}, + "call": { + "model": "%s", + "future_scalar": "ignored", + "future_object": {"nested": {"deep": 1}}, + "future_array": [{"a": 1}, {"b": 2}], + "usage": { + "input_tokens": 11, + "cache_read_input_tokens": 22, + "cache_creation_input_tokens": 33, + "cache_creation": { + "ephemeral_5m_input_tokens": 44, + "ephemeral_1h_input_tokens": 55, + "ephemeral_7d_input_tokens": 77 + }, + "output_tokens": 66, + "service_tier": "standard", + "inference_geo": "not_available" + }, + "config": { + "effort": "high", + "thinking_type": "adaptive", + "max_tokens": 64000, + "context_management": "clear_thinking_20251015", + "speed": "fast", + "future_knob": true + }, + "trigger": "subagent", + "trigger_detail": "Explore", + "turn_key": "unknown-fields-turnkey", + "parent_tool_use_id": "toolu_unknown_fields", + "spawn_depth": 2 + }, + "blocks": [ + {"category":"skills_loaded","side":"input","cache_status":"read","parent_category":"context","chars":100,"tool_name":"","tool_server":"","tool_use_id":"","resource":"dataviz","kind":"text","future_block_field":"ignored"} + ] + } + } + """ + .formatted(model)); + } + private static JsonNode spanCipxMetadata(String model, long input, long cacheRead, long cacheCreation, long cacheCreation5m, long cacheCreation1h, long output) { return JsonUtils.getJsonNodeFromString( @@ -861,6 +1088,13 @@ private record CipxSpendRow(String projectId, Long startMs, String model, Long u String triggerDetail, String turnKey, String parentToolUseId) { } + private record SentinelSpendRow(String workspaceId, String projectId, String traceId, String spanId, + Long startMs, String model, Long uInput, Long uCacheRead, Long uCacheCreation, Long uCacheCreation5m, + Long uCacheCreation1h, Long uOutput, String effort, String thinkingType, Long maxTokens, + String contextManagement, String speed, String trigger, String triggerDetail, String turnKey, + String parentToolUseId) { + } + private record CipxBlockRow(Integer blockIdx, String src, String category, String tier, String lane, String bdLane, String label, Integer isDefinition, Double alloc, String model, String speed, String side, From 79b112a00489227cea06357c7941d1957e573918 Mon Sep 17 00:00:00 2001 From: aadereiko Date: Thu, 20 Aug 2026 18:22:34 +0200 Subject: [PATCH 3/4] test(cipx-spends): await the block assertion; end migration with an empty line Both from baz review feedback on #7937. The unknown-field test awaited the spend row and then asserted the blocks outside the callback. CostIntelligenceIngestionListener subscribes to the spend insert and the block insert independently, so the spend row landing says nothing about whether the blocks have - a real flake. Moved inside the same await. Migration now ends with an empty line, per .agents/skills/opik-backend/ migrations.md ("Always end with empty line"). Note 000117 does not, so the convention is documented but not uniformly followed. Co-Authored-By: Claude Opus 5 (1M context) --- .../migrations/000118_add_trigger_to_cipx_spends.sql | 1 + .../v1/events/CostIntelligenceIngestionTest.java | 10 ++++++---- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql b/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql index 18afaaf2c74..01949053843 100644 --- a/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql +++ b/apps/opik-backend/src/main/resources/liquibase/db-app-analytics/migrations/000118_add_trigger_to_cipx_spends.sql @@ -28,3 +28,4 @@ ALTER TABLE ${ANALYTICS_DB_DATABASE_NAME}.cipx_spends ON CLUSTER '{cluster}' ADD COLUMN IF NOT EXISTS parent_tool_use_id String DEFAULT ''; --rollback ALTER TABLE ${ANALYTICS_DB_DATABASE_NAME}.cipx_spends ON CLUSTER '{cluster}' DROP COLUMN IF EXISTS `trigger`, DROP COLUMN IF EXISTS trigger_detail, DROP COLUMN IF EXISTS turn_key, DROP COLUMN IF EXISTS parent_tool_use_id; + diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java index fd1afcf2c46..d30378167b9 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java @@ -258,11 +258,13 @@ void unknownFieldsDoNotRejectTheRow() { assertThat(row.get().triggerDetail()).isEqualTo("Explore"); assertThat(row.get().turnKey()).isEqualTo("unknown-fields-turnkey"); assertThat(row.get().parentToolUseId()).isEqualTo("toolu_unknown_fields"); + // The block writer sees the same metadata, so an unknown field on a block must not + // drop the blocks either. Asserted inside the same await: the listener subscribes to + // the spend insert and the block insert independently + // (CostIntelligenceIngestionListener), so the spend row landing says nothing about + // whether the blocks have. + assertThat(getCipxBlocks(span.id(), ws.workspaceId())).isNotEmpty(); }); - - // The block writer sees the same metadata, so an unknown field on a block must not drop - // the blocks either. - assertThat(getCipxBlocks(span.id(), ws.workspaceId())).isNotEmpty(); } @Test From b0941b38600e110144beec9ae0c6e775150419d4 Mon Sep 17 00:00:00 2001 From: aadereiko Date: Thu, 20 Aug 2026 18:48:48 +0200 Subject: [PATCH 4/4] test(cipx-spends): finish the truncated comment on the block assertion From baz review feedback on #7937. Co-Authored-By: Claude Opus 5 (1M context) --- .../api/resources/v1/events/CostIntelligenceIngestionTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java index d30378167b9..44839653082 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/CostIntelligenceIngestionTest.java @@ -262,7 +262,7 @@ void unknownFieldsDoNotRejectTheRow() { // drop the blocks either. Asserted inside the same await: the listener subscribes to // the spend insert and the block insert independently // (CostIntelligenceIngestionListener), so the spend row landing says nothing about - // whether the blocks have. + // whether the blocks have landed. assertThat(getCipxBlocks(span.id(), ws.workspaceId())).isNotEmpty(); }); }