feat(ai-cost): ingest Claude Team vendor invoices

.github/workflows/e2e-bronze-to-api.yml

@@ -428,7 +428,10 @@ jobs:
           # sync, a silver-only contract. The loop below only ever yields
           # yaml-derived nodes, so they are listed per shard or they never run.
           case "$SHARD" in
-            ai) nodes+=("metrics/test_ai_seat_extra_usage_history.py") ;;
+            ai)
+              nodes+=("metrics/test_ai_seat_extra_usage_history.py")
+              nodes+=("metrics/test_ai_invoice_silver.py")
+              ;;
           esac
           while IFS= read -r stem; do
             nodes+=("metrics/test_fixtures.py::test_metric_smoke[${stem}]")

.trivyignore.yaml

@@ -18,3 +18,13 @@ misconfigurations:
       repository writable, so a baked-in UID breaks writes whenever it differs from
       the host's, taking the e2e gate with it. Revisit if the image is ever
       published or run anywhere other than a test job.
+  - id: AVD-DS-0026
+    paths:
+      - "src/ingestion/connectors/ai/claude-team-invoices/Dockerfile"
+    statement: >-
+      Not applicable. Docker polls a HEALTHCHECK against a long-running service;
+      this image runs one Airbyte protocol command and exits, and nothing polls
+      it — neither the job that calls `discover` nor the k8s Job that syncs. The
+      base image declares none either, so the instruction would report on a
+      process that has already finished. Revisit if this connector is ever run as
+      a service rather than as a command.

scripts/ci/components.py

@@ -175,6 +175,13 @@
         "cov_package": "source_bamboohr",
         "paths": ["src/ingestion/connectors/hr-directory/bamboohr"],
     },
+    {
+        "name": "claude-team-invoices",
+        "lang": "python",
+        "root": "src/ingestion/connectors/ai/claude-team-invoices",
+        "cov_package": "source_claude_team_invoices",
+        "paths": ["src/ingestion/connectors/ai/claude-team-invoices"],
+    },
     # Deploy-time ClickHouse schema tooling (the migration Job's Python half:
     # reconcile_bronze_schema, which heals warm-cluster bronze drift — #1991).
     # Owning the whole scripts/ tree means a connectors-ddl snapshot regen also

src/ingestion/.gitignore

@@ -4,6 +4,9 @@ connections/*.yaml
 
 # Generated schemas and catalogs (can be regenerated)
 connectors/**/schemas/
+# A CDK connector's schemas/ is not generated: the source package reads it at
+# discover time, so it has to ship with the code.
+!connectors/**/source_*/schemas/
 connectors/**/configured_catalog.json
 connectors/**/state.json
 

src/ingestion/connectors/ai/claude-team-invoices/Dockerfile

@@ -0,0 +1,18 @@
+FROM airbyte/python-connector-base:4.1.1
+
+COPY . /airbyte/integration_code
+RUN pip install /airbyte/integration_code
+
+# airbyte-cdk pins nltk==3.9.1 (fixable CVE-2025-14009, CVE-2026-0846,
+# CVE-2026-33231, CVE-2026-54293) but only uses it in the file-based
+# unstructured parser, which this connector never loads; defusedxml is
+# nltk 3.10's added runtime dependency. poetry/dulwich are base-image build
+# tooling unused at runtime (CVE-2026-34591, CVE-2026-42305).
+RUN pip install --no-cache-dir --no-deps nltk==3.10.0 defusedxml==0.7.1 \
+    && pip uninstall -y poetry dulwich
+
+RUN useradd --create-home --no-log-init --shell /bin/bash airbyte_user
+USER airbyte_user
+
+ENV AIRBYTE_ENTRYPOINT="source-claude-team-invoices"
+ENTRYPOINT ["source-claude-team-invoices"]

src/ingestion/connectors/ai/claude-team-invoices/README.md

@@ -0,0 +1,76 @@
+# Claude Team invoices
+
+The invoiced layer for Claude Team: what Anthropic actually billed, and the only
+place a seat's price appears.
+
+## Why a CDK connector
+
+The invoice list alone is declarative-shaped. Its line items are not: the
+claude.ai wrapper carries no invoice id and no lines, only a
+`hosted_invoice_url`, and reaching the lines takes a three-hop chain across two
+hosts where each hop's credential comes out of the previous hop's response.
+Same reasoning as github-copilot ADR-0001.
+
+## The chain
+
+| Step | Request | What it yields |
+|---|---|---|
+| 1 | `GET {proxy}/api/stripe/{org}/invoices` | the wrapper's rows: `total`, `total_excluding_tax`, `currency`, `status`, `num_seats`, `hosted_invoice_url` — no invoice id, no lines |
+| 2 | parse `https://invoice.stripe.com/i/{acct}/{token}?s=ap` | the account and token identifiers |
+| 3 | `GET invoicedata.stripe.com/hosted_invoice_page/{acct}/{token}` | `invoice_id` and a short-lived `ephemeral_key` |
+| 4 | `GET api.stripe.com/v1/invoices/{id}/lines?limit=100`, paginated on `has_more` | the full line set |
+
+Step 3 carries no header: the token inside its URL is what authorises it.
+Step 4 sends the ephemeral key, `Stripe-Version: 2026-06-24.dahlia` and
+`Stripe-Account: {acct}` on every line page. The key is short-lived and is
+never written to a record, a state message or a log line.
+
+Follow a URL inside the run that fetched it. Stripe expires a hosted invoice URL
+30 days after the due date, and claude.ai re-issues a fresh one on every list
+call — so storing a URL and following it later works in every test and then
+fails in production, oldest invoices first.
+
+**What the connector must be allowed to reach.** Steps 3 to 5 leave the cluster
+for `invoicedata.stripe.com` and `api.stripe.com`; where a `NetworkPolicy`
+governs egress it has to admit both. A blocked host surfaces as a chain step
+that did not complete, not as missing money.
+
+## What a line means
+
+Categories come from the Stripe parent, never from the description — a
+subscription-item parent is the recurring seat charge, an invoice-item parent is
+prepaid extra usage.
+
+A seat price is `hosted_invoice_unit_amount` on a non-proration subscription
+line. It is not `amount / quantity`: a mid-period seat change emits proration
+lines whose amounts cover part of a period, and dividing those yields a number
+that is not a price. It is also not the wrapper's `num_seats`, which may be
+absent and, where present, reports one line's quantity while an invoice can
+price several tiers.
+
+An invoice may price several tiers at once, each on its own line — so a seat
+price binds to a tier and reaches a person through `class_ai_overage.seat_tier`,
+never by dividing an invoice total.
+
+`period` dates a line to the window it charges for, which is not always the
+window the invoice was raised in. A line is filed by its period, never by the
+invoice date.
+
+Prepaid extra-usage purchases arrive as invoices of their own, carrying a single
+`overusage` line at `quantity: 1`. They are the invoiced-layer counterpart of
+the extra usage a seat later consumes, and they price no seat.
+
+## Degradation
+
+Every invoice emits its own row carrying its money and how far its chain got;
+lines are added beside it only when the chain completed. So a chain failure keeps
+the money on the ledger without a fabricated price, and the invoice keeps one row
+across attempts — a later run that enriches it replaces that row instead of
+adding its money a second time.
+
+`chain_status` distinguishes four outcomes: `ok`, `failed` (a hop answered
+badly), `unparsable_url` (a hosted URL was offered but no longer matches), and
+`no_hosted_url` (none was offered, as on a draft invoice). Only URLs that were
+offered count towards drift: if more than half of them fail to parse the run
+fails instead, because that is a format change, and a run of unpriced rows would
+read as the vendor having stopped charging for seats.

src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql

@@ -0,0 +1,104 @@
+-- depends_on: {{ ref('claude_team_invoices__bronze_promoted') }}
+-- Bronze → Silver: Claude Team vendor invoices → class_ai_invoice
+--
+-- Source: bronze_claude_team_invoices.claude_team_invoice_lines — one row per
+-- invoice plus one per invoice line, produced by the CDK connector that walks the
+-- claude.ai invoice wrapper and follows the Stripe hosted chain behind it.
+--
+-- This is the FIRST contributor to class_ai_invoice and therefore DEFINES its
+-- positional contract (consumed by union_by_tag('silver:class_ai_invoice')).
+-- Vendor-specific extras go into invoice_metrics_json, never new columns.
+--
+-- GRAIN: one row per (tenant, source, invoice) carrying that invoice's own
+-- money, plus one row per (tenant, source, invoice, line). Invoice money is on
+-- the invoice's row alone, so summing it needs no dedup and an invoice whose
+-- chain later completes replaces its own row instead of adding a second one.
+-- Aggregating to category is gold's job — a class that pre-aggregated would make
+-- the per-tier seat price unrecoverable.
+--
+-- UNITS: Stripe amounts are ALREADY minor units (cents), so they map straight
+-- through with NO ×100 — same as claude_team__ai_overage, unlike
+-- claude_team__ai_dev_usage.
+--
+-- PERIOD: dated by the window the line CHARGES for, not by the day the invoice
+-- was issued — a monthly invoice is raised at the period boundary and would
+-- otherwise land in the neighbouring month. Rows carrying no line fall back to
+-- the invoice date.
+-- STRATEGY: delete+insert, not append. An invoice's row is rewritten whenever its
+-- chain gets further, so the same unique_key arrives twice; appending would leave
+-- both versions standing until a background merge collapsed them, and the
+-- `unique` test reads without FINAL.
+{{ config(
+    materialized='incremental',
+    incremental_strategy='delete+insert',
+    unique_key='unique_key',
+    engine='ReplacingMergeTree(_version)',
+    order_by=['unique_key'],
+    on_schema_change='append_new_columns',
+    settings={'allow_nullable_key': 1},
+    schema='staging',
+    tags=['claude-team-invoices', 'silver:class_ai_invoice']
+) }}
+
+WITH latest_per_key AS (
+
+    -- Bronze is full-refresh+append: every sync re-emits every invoice and line
+    -- under the same unique_key, and an invoice's row changes when its chain
+    -- finally completes. Collapse to the freshest read per key.
+    SELECT *
+    FROM {{ source('bronze_claude_team_invoices', 'claude_team_invoice_lines') }}
+    WHERE unique_key IS NOT NULL
+      AND unique_key != ''
+    ORDER BY _airbyte_extracted_at DESC
+    LIMIT 1 BY unique_key
+
+)
+
+SELECT
+    tenant_id                                           AS insight_tenant_id,
+    source_id,
+    unique_key,
+    invoice_id,
+    line_id,
+    'claude'                                            AS tool,
+    invoice_status,
+    -- Chain outcome, carried so an unenriched invoice reads as unenriched
+    -- rather than as an invoice that happened to have no lines.
+    chain_status,
+    category,
+    tier_label,
+    toUInt8(coalesce(is_proration, false))              AS is_proration,
+    coalesce(currency, invoice_currency, 'usd')         AS currency,
+    toStartOfMonth(toDateTime(toInt64(coalesce(period_start_ts, invoice_created_ts, 0)))) AS period_month,
+    toInt64OrNull(toString(round(amount)))              AS amount_cents,
+    -- The per-seat price the vendor states on the line. NULL on prorations and
+    -- on extra usage — honest-NULL: neither prices a seat.
+    toInt64OrNull(toString(round(seat_unit_amount)))    AS seat_unit_cents,
+    toInt64OrNull(toString(round(quantity)))            AS seat_quantity,
+    toInt64OrNull(toString(round(invoice_total_excluding_tax))) AS invoice_net_cents,
+    -- Vendor extras kept out of the positional contract.
+    toJSONString(map(
+        'product_name',  ifNull(toString(product_name), ''),
+        'description',   ifNull(toString(description), ''),
+        'num_seats',     ifNull(toString(invoice_num_seats), ''),
+        'invoice_total', ifNull(toString(invoice_total), ''),
+        'period_end_ts', ifNull(toString(period_end_ts), '')
+    ))                                                  AS invoice_metrics_json,
+    'claude_team'                                       AS source,
+    data_source,
+    CAST(_airbyte_extracted_at AS Nullable(DateTime64(3))) AS collected_at,
+    toUnixTimestamp64Milli(_airbyte_extracted_at)          AS _version
+FROM latest_per_key
+{% if is_incremental() %}
+  -- A row only ever changes because a newer read produced it, so rows read since
+  -- the last build are the only ones that can carry anything new. The empty-table
+  -- guard mirrors the sibling models: over an empty `this` the max is the epoch
+  -- and every row would be filtered out.
+  WHERE (
+    (SELECT count() FROM {{ this }}) = 0
+    OR _airbyte_extracted_at > (
+        SELECT coalesce(max(collected_at), toDateTime64('1970-01-01 00:00:00', 3))
+        FROM {{ this }}
+    )
+  )
+{% endif %}

src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team_invoices__bronze_promoted.sql

@@ -0,0 +1,19 @@
+{# -------------------------------------------------------------------------
+   Bootstrap model for claude-team-invoices bronze → RMT promotion.
+
+   Counterpart of `claude_team__bronze_promoted` for the invoice connector,
+   which owns its own bronze namespace. See ADR-0002. The
+   `promote_bronze_to_rmt` macro is idempotent — already-RMT tables are
+   detected and skipped on subsequent runs.
+   ------------------------------------------------------------------------- #}
+
+-- @cpt-principle:cpt-dataflow-principle-promote-bronze:p1
+{{ config(
+    materialized='view',
+    schema='staging',
+    tags=['claude-team-invoices']
+) }}
+
+{% do promote_bronze_to_rmt(table='bronze_claude_team_invoices.claude_team_invoice_lines', order_by='unique_key') %}
+
+SELECT 1 AS promoted