From 3c7264dbf91637b8692f9ba6f852c955096f31fd Mon Sep 17 00:00:00 2001 From: Gregory Gogin Date: Mon, 17 Aug 2026 14:20:29 +0200 Subject: [PATCH 1/5] fix(ai-cost): make a changed seat-price type visible instead of absent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `seat_unit_amount` returns None for anything that is not an int, so if the hosted-invoice surface ever serialises the unit amount as a float or a digit string, every seat price in the run becomes NULL while `chain_status` still reads `ok` — indistinguishable from a vendor that prices no seats, and invisible to the coverage check, which only looks at rows that are not `ok`. A seat-pricing line whose amount arrives as another type now degrades its invoice to `failed`, which is the state the coverage check already watches and the remediation already explains. Absence keeps its own meaning: a subscription line the vendor left unpriced is a state this connector reports, not a shape it failed to read, so only a changed type degrades. The warning carries the offending type by name. A `StripeChainError` is authored here and names what the response got wrong, so its message is logged; anything else still contributes its type and status alone, because a request error stringifies its URL and the hosted-invoice hop carries the token in that URL. A bare `failed` row would otherwise send the operator to the egress and Stripe-Version checks the remediation names, neither of which is the fault. Signed-off-by: Gregory Gogin --- .../stripe_chain.py | 39 +++++++++++++++++-- .../tests/test_build_records.py | 28 +++++++++++++ .../tests/test_stripe_chain.py | 20 ++++++++++ 3 files changed, 83 insertions(+), 4 deletions(-) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py index 37aee0810..7327bea49 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py @@ -83,6 +83,11 @@ def is_proration(line: Mapping[str, Any]) -> bool: return bool(details.get("proration")) +def prices_a_seat(line: Mapping[str, Any]) -> bool: + """Whether this line is the kind that states a per-seat price at all.""" + return classify_line(line) == CATEGORY_SUBSCRIPTIONS and not is_proration(line) + + def seat_unit_amount(line: Mapping[str, Any]) -> int | None: """The per-seat price on a line, in minor units, or None when it has none. @@ -90,7 +95,7 @@ def seat_unit_amount(line: Mapping[str, Any]) -> int | None: extra usage, prorations, a subscription line the vendor left unpriced — yields None, which downstream renders as absence rather than as zero. """ - if classify_line(line) != CATEGORY_SUBSCRIPTIONS or is_proration(line): + if not prices_a_seat(line): return None amount = line.get("hosted_invoice_unit_amount") # `bool` is an `int` in Python, and a boolean here would price a seat at 1. @@ -99,6 +104,26 @@ def seat_unit_amount(line: Mapping[str, Any]) -> int | None: return int(amount) +def unreadable_seat_prices(lines: Sequence[Mapping[str, Any]]) -> list[str]: + """Type names found where a seat-pricing line should state an integer amount. + + `None` is not one of them: a subscription line the vendor left unpriced is a + state this connector reports rather than a shape it failed to read. Anything + else — a float, a digit string — is the vendor having changed the field's type, + and returning absence for it would put every seat price at NULL while the + chain still read as complete. + """ + offenders = [] + for line in lines: + if not prices_a_seat(line): + continue + amount = line.get("hosted_invoice_unit_amount") + if amount is None or (isinstance(amount, int) and not isinstance(amount, bool)): + continue + offenders.append(type(amount).__name__) + return sorted(set(offenders)) + + def shape_line(line: Mapping[str, Any], invoice_key: str) -> dict[str, Any]: """Project one Stripe line onto the columns bronze keeps. @@ -291,14 +316,20 @@ def build_records( continue try: invoice_id, lines = fetch_lines(ref.acct, ref.token) + unreadable = unreadable_seat_prices(lines) + if unreadable: + raise StripeChainError(f"hosted_invoice_unit_amount arrived as {', '.join(unreadable)}, not an integer") except Exception as error: # noqa: BLE001 - one invoice must not end the run # A gap in pricing, not a reason to lose the invoice or fail the sync. - # The type and status only: a request error stringifies its URL, and - # the hosted-invoice hop carries the token in that URL. + # Only a StripeChainError carries its message: those are authored here + # and name what the response got wrong, which is the whole diagnostic + # for a shape change. Anything else contributes its type and status + # alone — a request error stringifies its URL, and the hosted-invoice + # hop carries the token in that URL. logger.warning( "stripe chain failed for the invoice created at %s: %s (HTTP %s)", invoice.get("created_ts"), - type(error).__name__, + str(error) if isinstance(error, StripeChainError) else type(error).__name__, getattr(getattr(error, "response", None), "status_code", "n/a"), ) yield invoice_row(invoice, CHAIN_FAILED, None) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py index 72f4f0abe..9a00f15b9 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py @@ -130,6 +130,34 @@ def flaky(acct, token): assert "stripe chain failed" in caplog.text +def test_a_seat_price_of_an_unexpected_type_degrades_the_invoice_visibly(caplog: pytest.LogCaptureFixture) -> None: + """Returning absence instead would put every seat price at NULL while the + chain still read as complete — indistinguishable from a vendor pricing no seats.""" + + def lines_with_a_string_price(acct, token): + return "in_1ABC", [dict(SUBSCRIPTION_LINE, hosted_invoice_unit_amount="1000")] + + with caplog.at_level(logging.WARNING): + records = list(build_records([invoice()], lines_with_a_string_price)) + + assert [r["chain_status"] for r in records] == [CHAIN_FAILED] + assert records[0]["invoice_total_excluding_tax"] == 3000, "the ledger survives" + # The type it arrived as, by name — a `failed` row alone would send the + # operator to the egress and Stripe-Version checks the remediation names. + assert "hosted_invoice_unit_amount arrived as str" in caplog.text + + +def test_a_seat_line_the_vendor_left_unpriced_does_not_degrade_the_invoice() -> None: + """Absence is a state this connector reports; only a changed type is a fault.""" + + def lines_unpriced(acct, token): + return "in_1ABC", [dict(SUBSCRIPTION_LINE, hosted_invoice_unit_amount=None)] + + records = list(build_records([invoice()], lines_unpriced)) + assert [r["chain_status"] for r in records] == [CHAIN_OK, CHAIN_OK] + assert [r["seat_unit_amount"] for r in records] == [None, None] + + def test_a_run_of_unparsable_urls_fails_instead_of_writing_priceless_rows() -> None: invoices = [invoice(url="https://elsewhere.example/x") for _ in range(3)] + [invoice()] with pytest.raises(UrlFormatDrift) as raised: diff --git a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stripe_chain.py b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stripe_chain.py index 6677a7565..4132237bb 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stripe_chain.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stripe_chain.py @@ -19,6 +19,7 @@ read_bootstrap, seat_unit_amount, shape_line, + unreadable_seat_prices, ) SUBSCRIPTION_LINE = { @@ -113,6 +114,25 @@ def test_a_boolean_unit_amount_is_absence_not_a_price(value: bool) -> None: assert seat_unit_amount(dict(SUBSCRIPTION_LINE, hosted_invoice_unit_amount=value)) is None +@pytest.mark.parametrize("value", [2500.0, "2500", True, [], {}]) +def test_a_unit_amount_of_another_type_is_reported_as_unreadable(value: object) -> None: + """Absence and a changed type look the same downstream; only one is a fault.""" + lines = [dict(SUBSCRIPTION_LINE, hosted_invoice_unit_amount=value)] + assert unreadable_seat_prices(lines) == [type(value).__name__] + + +def test_an_unpriced_seat_line_is_absence_not_an_unreadable_type() -> None: + """The vendor leaving a subscription line unpriced is a state we report.""" + assert unreadable_seat_prices([dict(SUBSCRIPTION_LINE, hosted_invoice_unit_amount=None)]) == [] + assert unreadable_seat_prices([SUBSCRIPTION_LINE]) == [] + + +def test_only_seat_pricing_lines_are_judged_on_their_unit_amount() -> None: + """Extra usage and prorations never state a seat price, so their type is not ours.""" + assert unreadable_seat_prices([dict(EXTRA_USAGE_LINE, hosted_invoice_unit_amount="2000")]) == [] + assert unreadable_seat_prices([dict(PRORATION_CREDIT, hosted_invoice_unit_amount="1500")]) == [] + + def test_proration_is_read_from_the_structural_flag() -> None: assert is_proration(PRORATION_CREDIT) assert not is_proration(SUBSCRIPTION_LINE) From d3db5c6608d0585d9fc490cc8a95829bb399f468 Mon Sep 17 00:00:00 2001 From: Gregory Gogin Date: Mon, 17 Aug 2026 18:16:54 +0200 Subject: [PATCH 2/5] fix(ai-cost): do not emit drafts, which carry no stable identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A draft has no hosted invoice URL — the vendor issues one only at finalisation — so there is nothing invariant to identify it by, and its row can only be keyed on what the wrapper reports: created timestamp, payment intent, total and due date. Those are exactly the fields that move when the draft is finalised. The payment intent is created at that moment and a draft's total can still change until then, so the same invoice would be keyed one way as a draft and another once finalised: the draft's row stays in the class beside the real one, and a sum over `invoice_net_cents` counts that invoice twice. Nothing would surface it, because a draft carries no lines and the coverage check only looks for missing lines. Skipping them is the cheaper guard than keeping their money off the row, and it is also the truthful one: a draft is not invoiced, and this is the invoiced layer. Its total is provisional by definition. Two things fall out. `no_hosted_url` now means only what it should — the vendor offered no hosted URL for an invoice it has finalised, which is a vendor-side change rather than a legitimate state — so the coverage check no longer needs to exclude drafts and reports every status again. And drafts stay out of the drift ratio, where two of them beside one malformed URL could have tipped a healthy run into refusing to write. The count of skipped drafts is logged: an operator comparing the vendor's list against ours should not have to guess why the totals differ. Signed-off-by: Gregory Gogin --- .../ai/claude-team-invoices/README.md | 13 +++++--- .../ai/claude-team-invoices/dbt/schema.yml | 7 +++-- .../stripe_chain.py | 26 +++++++++++++++- .../tests/test_build_records.py | 31 +++++++++++++++++++ .../ai/assert_ai_invoice_lines_enriched.sql | 10 +++--- .../e2e/metrics/test_ai_invoice_silver.py | 7 +++-- 6 files changed, 77 insertions(+), 17 deletions(-) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/README.md b/src/ingestion/connectors/ai/claude-team-invoices/README.md index 8707c8d17..41a15b647 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/README.md +++ b/src/ingestion/connectors/ai/claude-team-invoices/README.md @@ -68,9 +68,14 @@ the money on the ledger without a fabricated price, and the invoice keeps one ro across attempts — a later run that enriches it replaces that row instead of adding its money a second time. +A draft is not invoiced and is skipped entirely. Its total can still change and +it carries no payment intent, and both are part of the key an invoice's row is +identified by — so emitting one would leave the draft's copy standing beside the +finalised invoice, counting that money twice. + `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. +`no_hosted_url` (none was offered for an invoice the vendor has finalised). 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. diff --git a/src/ingestion/connectors/ai/claude-team-invoices/dbt/schema.yml b/src/ingestion/connectors/ai/claude-team-invoices/dbt/schema.yml index 0746016ee..58e69ca10 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/dbt/schema.yml +++ b/src/ingestion/connectors/ai/claude-team-invoices/dbt/schema.yml @@ -65,9 +65,10 @@ models: How far the Stripe chain got for this invoice: 'ok' (lines retrieved), 'failed' (a hop raised), 'unparsable_url' (the vendor offered a hosted URL that no longer matches the expected form), 'no_hosted_url' (the - vendor offered none, which a draft invoice legitimately does). Rows - other than 'ok' are the invoice's own row with no line, so an - unenriched invoice is visible rather than silently absent. + vendor offered none for an invoice it has finalised). Rows other than + 'ok' are the invoice's own row with no line, so an unenriched invoice is + visible rather than silently absent. Drafts never reach this class: they + are not invoiced yet, so the connector does not emit them. tests: - not_null - accepted_values: diff --git a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py index 7327bea49..4f9b9cd41 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py @@ -55,6 +55,21 @@ def parse_hosted_invoice_url(url: str | None) -> HostedRef | None: return HostedRef(match.group(1), match.group(2)) if match else None +INVOICE_STATUS_DRAFT = "draft" + + +def is_draft(invoice: Mapping[str, Any]) -> bool: + """Whether the vendor has yet to finalise this invoice. + + A draft is not invoiced: it carries no payment intent and a total that can + still change, and those are two of the five fields an invoice's row is keyed + on. Emitting one would put provisional money on a row whose key changes the + moment the invoice is finalised, leaving the draft's copy standing beside the + real one and counting that invoice's money twice. + """ + return str(invoice.get("status") or "").lower() == INVOICE_STATUS_DRAFT + + def classify_line(line: Mapping[str, Any]) -> str: """Categorise one invoice line by its Stripe parent, never by its text. @@ -291,6 +306,8 @@ def build_records( money sits on exactly one row whatever happens, and a run that enriches an invoice replaces the row an earlier unenriched run wrote for it. + A draft is not one of them: it is not invoiced yet, and it is skipped entirely. + Four outcomes per invoice, and one for the run: * the wrapper offered no URL -> `no_hosted_url` * the URL did not parse -> `unparsable_url` @@ -299,7 +316,14 @@ def build_records( and if more than `drift_ratio` of the URLs the vendor did offer failed to parse, the run raises rather than writing rows that carry money but no prices. """ - parsed = [(inv, parse_hosted_invoice_url(inv.get("hosted_invoice_url"))) for inv in invoices] + invoiced = [inv for inv in invoices if not is_draft(inv)] + skipped = len(invoices) - len(invoiced) + if skipped: + # Said out loud: an operator comparing the vendor's list against ours + # should not have to guess why a count differs. + logger.info("skipped %s draft invoice(s): not invoiced yet, so not on the ledger", skipped) + + parsed = [(inv, parse_hosted_invoice_url(inv.get("hosted_invoice_url"))) for inv in invoiced] offered = [inv for inv, _ in parsed if inv.get("hosted_invoice_url")] malformed = sum(1 for inv, ref in parsed if ref is None and inv.get("hosted_invoice_url")) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py index 9a00f15b9..d5056ee6a 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py @@ -158,6 +158,37 @@ def lines_unpriced(acct, token): assert [r["seat_unit_amount"] for r in records] == [None, None] +def test_a_draft_is_not_emitted_at_all(caplog: pytest.LogCaptureFixture) -> None: + """Its total can still change and it has no payment intent — two of the fields + an invoice's row is keyed on — so it would duplicate on finalisation.""" + with caplog.at_level(logging.INFO): + records = list(build_records([invoice(status="draft", payment_intent=None, url=None)], lines_ok)) + + assert records == [] + assert "skipped 1 draft invoice(s)" in caplog.text + + +def test_a_draft_does_not_hide_the_invoices_beside_it() -> None: + drafts = [invoice(status="draft", payment_intent=None, url=None) for _ in range(2)] + records = list(build_records([*drafts, invoice()], lines_ok)) + assert [r["chain_status"] for r in records] == [CHAIN_OK, CHAIN_OK, CHAIN_OK] + assert [r["line_id"] for r in records] == [None, "il_standard", "il_prepaid"] + + +def test_a_draft_stays_out_of_the_drift_ratio() -> None: + """Counted in, two drafts beside one bad URL would tip a healthy run into drift.""" + drafts = [invoice(status="draft", url=None) for _ in range(2)] + records = list(build_records([*drafts, invoice(url="bad"), invoice()], lines_ok)) + assert sum(1 for r in records if r["chain_status"] == CHAIN_UNPARSABLE) == 1 + assert sum(1 for r in records if r["chain_status"] == CHAIN_OK) == 3 + + +def test_a_finalised_invoice_without_a_url_is_still_reported() -> None: + """The status exists for this case now that drafts never reach it.""" + records = list(build_records([invoice(url=None)], lines_ok)) + assert [r["chain_status"] for r in records] == [CHAIN_NO_URL] + + def test_a_run_of_unparsable_urls_fails_instead_of_writing_priceless_rows() -> None: invoices = [invoice(url="https://elsewhere.example/x") for _ in range(3)] + [invoice()] with pytest.raises(UrlFormatDrift) as raised: diff --git a/src/ingestion/dbt/tests/ai/assert_ai_invoice_lines_enriched.sql b/src/ingestion/dbt/tests/ai/assert_ai_invoice_lines_enriched.sql index 0b20847f7..6b65205ae 100644 --- a/src/ingestion/dbt/tests/ai/assert_ai_invoice_lines_enriched.sql +++ b/src/ingestion/dbt/tests/ai/assert_ai_invoice_lines_enriched.sql @@ -7,7 +7,7 @@ 'domain': 'ai', 'category': 'coverage', 'tier': 'error', - 'remediation': 'A row here is an invoice whose Stripe hosted chain did not complete, so its money is on the ledger with no per-seat price behind it and ai.seat_cost is missing that tier for that month. chain_status says why: no_hosted_url means the vendor offered no hosted URL at all, which a draft invoice legitimately does and which resolves when the invoice is finalised; unparsable_url means a URL was offered but did not match the expected form, which for more than an odd invoice means the format changed and the connector needs updating; failed means a hop answered badly — check egress to invoicedata.stripe.com and api.stripe.com, and that the pinned Stripe-Version is still accepted. Re-running the connector re-follows a freshly issued URL, and the invoice keeps one row across attempts, so a transient failure clears itself on the next sync.' + 'remediation': 'A row here is an invoice whose Stripe hosted chain did not complete, so its money is on the ledger with no per-seat price behind it and ai.seat_cost is missing that tier for that month. chain_status says why: no_hosted_url means the vendor offered no hosted URL for an invoice it has finalised, which is a vendor-side change rather than a fault of ours, since drafts are never emitted; unparsable_url means a URL was offered but did not match the expected form, which for more than an odd invoice means the format changed and the connector needs updating; failed means a hop answered badly — check egress to invoicedata.stripe.com and api.stripe.com, and that the pinned Stripe-Version is still accepted. Re-running the connector re-follows a freshly issued URL, and the invoice keeps one row across attempts, so a transient failure clears itself on the next sync.' } ) }} {#- Bounded to the last three billing months on purpose. An invoice that never @@ -16,10 +16,9 @@ actionable ones: the wrapper re-issues a fresh URL on every run, so a recent failure is one the next sync can still fix. - Drafts are excluded, and only drafts: a draft invoice has no hosted URL and no - final money yet, so reporting it is noise. A FINALISED invoice without one is - kept — that is what a vendor-side change would look like, and staying silent - about it is the silent emptiness this connector exists to avoid. -#} + No status is excluded: the connector does not emit drafts, so everything that + reaches this class is an invoice the vendor has finalised, and any of those + without lines is worth reporting. -#} SELECT insight_tenant_id, @@ -30,5 +29,4 @@ SELECT invoice_net_cents FROM {{ ref('class_ai_invoice') }} FINAL WHERE chain_status != 'ok' - AND ifNull(invoice_status, '') != 'draft' AND period_month >= toStartOfMonth(today()) - INTERVAL 2 MONTH diff --git a/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py b/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py index 5b4e4a632..56ed49241 100644 --- a/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py +++ b/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py @@ -197,8 +197,9 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at # An invoice whose chain never completed: the ledger survives, the price does # not, and with no line there is only the raise date to file it by. _invoice_row("pi_broken", 999, 999, chain_status="failed", invoice_currency=None), - # A draft the vendor offered no hosted URL for — absence, not a format change. - _invoice_row("pi_draft", 750, 750, chain_status="no_hosted_url", invoice_status="draft"), + # A finalised invoice the vendor offered no hosted URL for. Not a draft: the + # connector skips those, so one could never reach bronze in the first place. + _invoice_row("pi_no_url", 750, 750, chain_status="no_hosted_url", invoice_status="open"), ] COLUMNS = [ @@ -300,7 +301,7 @@ def test_a_failed_chain_keeps_the_invoice_and_no_line(invoice_silver): def test_an_invoice_with_no_hosted_url_reaches_the_class_as_its_own_state(invoice_silver): - """A draft carries no URL; reading that as a format change would fail the sync.""" + """Reading a missing URL as a format change would fail the whole sync instead.""" row = _by_status(invoice_silver, "no_hosted_url") assert row["invoice_net_cents"] == 750 assert row["line_id"] is None and row["invoice_id"] is None From e2eb1799b79350b34e72d23dc650165242fe1438 Mon Sep 17 00:00:00 2001 From: Gregory Gogin Date: Tue, 18 Aug 2026 16:52:23 +0200 Subject: [PATCH 3/5] fix(ai-cost): pace the Stripe chain between invoices MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run walks every invoice the organisation has, and each one costs two requests to `invoicedata.stripe.com` — the host this connector's own README names as its known risk, since it appears in no Stripe documentation and carries no contract. Nothing paced them. A quarter-second sits between consecutive chains, and only between them: the hops inside one chain stay back to back, because the ephemeral key that authorises the last one is short-lived. The reference implementation paces the same boundary by the same amount. Signed-off-by: Gregory Gogin --- .../source_claude_team_invoices/streams.py | 11 ++++++++++- .../tests/test_stream_over_http.py | 19 +++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/streams.py b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/streams.py index 192b7f2bc..7fc65a4b3 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/streams.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/streams.py @@ -30,6 +30,7 @@ from __future__ import annotations +import time from collections.abc import Iterable, Mapping, MutableMapping, Sequence from datetime import UTC, datetime from typing import Any @@ -49,6 +50,9 @@ MAX_PAGES = 50 MAX_LINE_PAGES = 20 TIMEOUT = 30 +# Paced between invoices, not inside one: the bootstrap host is undocumented and +# a run walks every invoice, so the chain is the only part whose rate we choose. +CHAIN_DELAY_SECS = 0.25 def _now_iso() -> str: @@ -56,7 +60,7 @@ def _now_iso() -> str: class InvoiceLines(Stream): - """`claude_team_invoice_lines` — one record per invoice line.""" + """`claude_team_invoice_lines` — one record per invoice, plus one per line.""" primary_key = "unique_key" @@ -67,6 +71,7 @@ def __init__(self, config: Mapping[str, Any]) -> None: self._tenant_id = config["insight_tenant_id"] self._source_id = config["insight_source_id"] self._session = requests.Session() + self._chained = False @property def name(self) -> str: @@ -102,6 +107,10 @@ def _walk_invoices(self) -> list[Mapping[str, Any]]: def _fetch_lines(self, acct: str, token: str) -> tuple[str, Sequence[Mapping[str, Any]]]: """Run the Stripe hops and return the invoice id with its full line set.""" + if self._chained: + time.sleep(CHAIN_DELAY_SECS) + self._chained = True + bootstrap = self._get_json(f"{BOOTSTRAP_HOST}/hosted_invoice_page/{acct}/{token}") invoice_id, ephemeral_key = read_bootstrap(bootstrap) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py index 32c590d0a..e68d7f331 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py @@ -16,6 +16,7 @@ import pytest import requests_mock as rm_module +from source_claude_team_invoices import streams as streams_module from source_claude_team_invoices.streams import BOOTSTRAP_HOST, STRIPE_API, STRIPE_VERSION, InvoiceLines CONFIG = { @@ -147,6 +148,24 @@ def test_line_pages_are_walked_until_the_endpoint_stops_offering_more( assert "starting_after=il_first" in line_requests[1].url, "the second page resumes after the first page's last id" +def test_the_chain_is_paced_between_invoices_but_not_before_the_first( + stream: InvoiceLines, http: rm_module.Mocker, monkeypatch: pytest.MonkeyPatch +) -> None: + """The bootstrap host is undocumented and a run walks the whole history, so the + chain is the one part whose request rate is ours to choose.""" + slept: list[float] = [] + monkeypatch.setattr(streams_module.time, "sleep", slept.append) + + second = wrapper_invoice(payment_intent="pi_SECOND") + http.get(INVOICES_URL, json={"invoices": [wrapper_invoice(), second], "next_page": None}) + http.get(BOOTSTRAP_URL, json={"invoice_id": INVOICE_ID, "ephemeral_key": EPHEMERAL_KEY}) + http.get(LINES_URL, json={"data": [subscription_line()], "has_more": False}) + + list(stream.read_records(sync_mode="full_refresh")) + + assert slept == [streams_module.CHAIN_DELAY_SECS], "two invoices, one pause, none before the first" + + def test_a_hop_that_fails_leaves_a_gap_and_does_not_end_the_run( stream: InvoiceLines, http: rm_module.Mocker, caplog: pytest.LogCaptureFixture ) -> None: From a2868e837eea1122c52bcccccb8ddfbda70223de Mon Sep 17 00:00:00 2001 From: Gregory Gogin Date: Wed, 19 Aug 2026 09:43:42 +0200 Subject: [PATCH 4/5] fix(ai-cost): key an invoice on the identity decoded from its hosted URL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An invoice's own row was keyed on what the wrapper reports — created timestamp, payment intent, total and due date. Everything but the first can move over an invoice's life, and when the key moves the old row stays: silver has no deletion path, because class_ai_invoice filters its input with `_version > max(_version)` and delete+insert deletes only keys present in the incoming set. The invoice's money is then counted twice, and nothing surfaces it. The hosted invoice URL carries an identity that does not move. Its `live_…`/`test_…` segment is base64 of `acct_,_,`, and only the trailing field is re-issued on every list call. `stable_invoice_ref` decodes the first two into `invoice_ref`, `invoice_identity` puts it on every row of an invoice, and `unique_key_parts` keys on it — falling back to the wrapper's fields only where the vendor offered no URL and there is nothing else to key on. The ref reaches the class through `invoice_metrics_json` rather than as a column of its own: the staging model's header reserves the class contract for facts every contributor can supply, and vendor extras go into the JSON. A run whose URLs stop decoding is refused before the first chain call. Re-keying a whole run onto the mutable fields would write a second copy of every invoice that already has a row, and nothing downstream can delete the originals — the same reasoning as the existing unparsable-URL guard, at the same ratio. Two unit premises stopped being true and are replaced by the stronger claim they were reaching for: comparing a with-URL failure to a without-URL one no longer describes one invoice, and two invoices issued in one second are separated by their identities rather than by their totals. The e2e recovery fixture keyed its failed sync on the fallback while the recovered sync keyed on the identity, so the pair had stopped describing one invoice — a chain only fails once the URL has decoded, so both syncs carry the ref. The second-instance fixture gets one for the same reason: left without, it would exercise the fallback while every other invoice beside it exercises the identity, and nothing would say so. The bronze column needs no migration: reconcile_bronze_schema.py derives its ADD COLUMN statements from the snapshot and heals warm tables before the migrations run. Signed-off-by: Gregory Gogin --- .../dbt/claude_team__ai_invoice.sql | 1 + .../schemas/claude_team_invoice_lines.json | 1 + .../stripe_chain.py | 74 +++++++++++++++--- .../tests/test_build_records.py | 46 +++++++++-- .../tests/test_stream_over_http.py | 7 +- .../connectors-ddl/claude-team-invoices.sql | 1 + ...am_invoices.claude_team_invoice_lines.yaml | 1 + .../e2e/metrics/test_ai_invoice_silver.py | 76 +++++++++++++++---- 8 files changed, 173 insertions(+), 34 deletions(-) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql b/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql index 69372ef77..1db4f4009 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql +++ b/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql @@ -80,6 +80,7 @@ SELECT toJSONString(map( 'product_name', ifNull(toString(product_name), ''), 'description', ifNull(toString(description), ''), + 'invoice_ref', ifNull(toString(invoice_ref), ''), 'num_seats', ifNull(toString(invoice_num_seats), ''), 'invoice_total', ifNull(toString(invoice_total), ''), 'period_end_ts', ifNull(toString(period_end_ts), '') diff --git a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/schemas/claude_team_invoice_lines.json b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/schemas/claude_team_invoice_lines.json index 0e7911f88..2763029e3 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/schemas/claude_team_invoice_lines.json +++ b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/schemas/claude_team_invoice_lines.json @@ -10,6 +10,7 @@ "data_source": { "type": ["string", "null"] }, "collected_at": { "type": ["string", "null"] }, "chain_status": { "type": ["string", "null"] }, + "invoice_ref": { "type": ["string", "null"] }, "invoice_id": { "type": ["string", "null"] }, "invoice_status": { "type": ["string", "null"] }, "invoice_created_ts": { "type": ["number", "null"] }, diff --git a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py index 4f9b9cd41..304423adb 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/source_claude_team_invoices/stripe_chain.py @@ -10,6 +10,7 @@ from __future__ import annotations +import base64 import logging import re from collections.abc import Callable, Iterator, Mapping, Sequence @@ -23,7 +24,7 @@ # part of the pattern on purpose: these two segments are interpolated into a # request URL, so a link pointing anywhere else must not contribute them. _HOSTED_URL = re.compile( - r"^https://invoice\.stripe\.com/i/(acct_[A-Za-z0-9_-]+)/((?:test|live)_[A-Za-z0-9_-]+)(?:[?#]|$)" + r"^https://invoice\.stripe\.com/i/(acct_[A-Za-z0-9_-]+)/((?:test|live)_([A-Za-z0-9_-]+))(?:[?#]|$)" ) CATEGORY_SUBSCRIPTIONS = "subscriptions" @@ -61,15 +62,47 @@ def parse_hosted_invoice_url(url: str | None) -> HostedRef | None: def is_draft(invoice: Mapping[str, Any]) -> bool: """Whether the vendor has yet to finalise this invoice. - A draft is not invoiced: it carries no payment intent and a total that can - still change, and those are two of the five fields an invoice's row is keyed - on. Emitting one would put provisional money on a row whose key changes the + A draft is not invoiced, and it carries no hosted URL to take an identity + from, so its row could only key on what the wrapper reports — including the + payment intent it does not have yet and a total that can still change. + Emitting one would put provisional money on a row whose key changes the moment the invoice is finalised, leaving the draft's copy standing beside the real one and counting that invoice's money twice. """ return str(invoice.get("status") or "").lower() == INVOICE_STATUS_DRAFT +def stable_invoice_ref(url: str | None) -> str | None: + """The invoice's own identity, decoded out of its hosted URL. + + The URL's `live_…`/`test_…` segment is base64 of + `acct_,_,`, and the + vendor regenerates that trailing part on every list call — so only the first + two fields identify the invoice. Everything the wrapper reports beside the URL + either changes over an invoice's life (its total, its payment intent) or is + shared across a batch issued in the same second, which is why this is the + identity to key on and those are only a fallback. + + Returns None when no URL was offered or it does not decode; the caller then + falls back to what the wrapper reports. + """ + if not url: + return None + match = _HOSTED_URL.match(url) + if not match: + return None + try: + segment = match.group(3) + decoded = base64.urlsafe_b64decode(segment + "=" * (-len(segment) % 4)).decode("utf-8") + except (ValueError, UnicodeDecodeError): + return None + + fields = decoded.split(",") + if len(fields) < 2 or not fields[0].startswith("acct_") or not fields[1]: + return None + return f"{fields[0]},{fields[1]}" + + def classify_line(line: Mapping[str, Any]) -> str: """Categorise one invoice line by its Stripe parent, never by its text. @@ -218,6 +251,7 @@ def invoice_identity(invoice: Mapping[str, Any]) -> dict[str, Any]: "invoice_status": invoice.get("status"), "invoice_created_ts": invoice.get("created_ts"), "invoice_currency": invoice.get("currency"), + "invoice_ref": stable_invoice_ref(invoice.get("hosted_invoice_url")), } @@ -333,6 +367,17 @@ def build_records( "has almost certainly changed; refusing to write a run of unpriced rows" ) + # A URL can match the shape and still carry no decodable identity. Refuse before + # the first chain call: falling back to the wrapper's mutable fields for a whole + # run would key second copies of invoices that already have rows, and nothing + # downstream can delete the originals. + undecodable = sum(1 for inv in offered if stable_invoice_ref(inv.get("hosted_invoice_url")) is None) + if offered and undecodable > len(offered) * drift_ratio: + raise UrlFormatDrift( + f"{undecodable} of {len(offered)} hosted invoice URLs carry no decodable invoice " + "identity; refusing to re-key a whole run onto the wrapper's mutable fields" + ) + for invoice, ref in parsed: if ref is None: absent = not invoice.get("hosted_invoice_url") @@ -368,17 +413,22 @@ def unique_key_parts(record: Mapping[str, Any]) -> tuple[Any, ...]: """The natural key of a record, which differs by what the row carries. A line is identified by Stripe's own ids. An invoice's own row has no line - id, and no invoice id at all until its chain completes, so it keys on what - the wrapper reports on every run — which is what lets an enriched run replace - the row an unenriched one wrote instead of adding a second one beside it. - The chain outcome is deliberately NOT part of that key: an invoice that fails - one way and then another must stay one row. The amount and due date join it - because a payment intent is absent on some invoices, and creation timestamps - collide across a batch issued at once — two invoices sharing one key would - leave only one row. + id, and no invoice id at all until its chain completes, so it keys on the + identity decoded out of its hosted URL — which reads the same on every run + and is what lets an enriched run replace the row an unenriched one wrote + instead of adding a second one beside it. The chain outcome is deliberately + NOT part of that key: an invoice that fails one way and then another must + stay one row. + + With no URL there is no identity, and the fallback is what the wrapper + reports. The amount and due date join it there because a payment intent is + absent on some invoices, and creation timestamps collide across a batch + issued at once — two invoices sharing one key would leave only one row. """ if record.get("line_id"): return (record.get("invoice_id"), record.get("line_id")) + if record.get("invoice_ref"): + return ("invoice", record["invoice_ref"]) return ( "invoice", record.get("invoice_created_ts"), diff --git a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py index d5056ee6a..f4ea9661e 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_build_records.py @@ -5,6 +5,7 @@ a network, a cluster or the CDK to verify. """ +import base64 import logging from collections.abc import Mapping, Sequence from typing import Any @@ -22,12 +23,25 @@ ) from tests.test_stripe_chain import EXTRA_USAGE_LINE, SUBSCRIPTION_LINE -GOOD_URL = "https://invoice.stripe.com/i/acct_1ABC/live_TOKEN?s=ap" -FAILING_TOKEN = "live_FAILS" -FAILING_URL = f"https://invoice.stripe.com/i/acct_1ABC/{FAILING_TOKEN}?s=ap" EPHEMERAL_KEY = "ek_live_super_secret_value" +def _hosted(entity: str, nonce: str = "n1", acct: str = "acct_1ABC") -> tuple[str, str]: + """A hosted URL shaped the way the vendor's are, plus its path segment. + + The segment is base64 of `,_,`; the vendor regenerates + that trailing part on every list call, so two calls for one invoice differ in + the URL and still decode to the same identity. + """ + payload = f"{acct},_{entity},{nonce}" + segment = base64.urlsafe_b64encode(payload.encode()).decode().rstrip("=") + return f"https://invoice.stripe.com/i/{acct}/live_{segment}?s=ap", f"live_{segment}" + + +GOOD_URL, _GOOD_TOKEN = _hosted("ent1ABC") +FAILING_URL, FAILING_TOKEN = _hosted("entFAILS") + + def invoice(url: str | None = GOOD_URL, **over: Any) -> dict[str, Any]: base = { "hosted_invoice_url": url, @@ -259,11 +273,23 @@ def test_an_invoice_keeps_one_key_across_a_failure_and_a_later_recovery() -> Non assert unenriched["invoice_id"] is None and enriched["invoice_id"] == "in_1ABC" -def test_an_invoice_failing_two_different_ways_stays_one_key() -> None: - """The outcome is not part of the key, so two attempts cannot become two rows.""" - failed = next(iter(build_records([invoice()], lines_raise))) +def test_two_scrapes_of_one_invoice_share_a_key_though_its_url_rotated() -> None: + """The vendor re-issues a fresh URL on every list call; only the identity + inside it is the invoice, so that is what the key is built from.""" + first, _ = _hosted("ent1ABC", nonce="1756771200-n1") + later, _ = _hosted("ent1ABC", nonce="1756800000-n2") + assert first != later, "the fixture must actually rotate the URL" + + a = next(iter(build_records([invoice(url=first)], lines_raise))) + b = next(iter(build_records([invoice(url=later)], lines_raise))) + assert unique_key_parts(a) == unique_key_parts(b) == ("invoice", "acct_1ABC,_ent1ABC") + + +def test_an_invoice_whose_url_carries_no_identity_falls_back_to_the_wrapper() -> None: + """A documented limit, not a bug: with no identity the invoice cannot be tied + to its own enriched row, so the mutable fields are all that is left.""" without_url = next(iter(build_records([invoice(url=None)], lines_ok))) - assert unique_key_parts(failed) == unique_key_parts(without_url) + assert unique_key_parts(without_url) == ("invoice", 1756771200, "pi_1", 3300, None) def test_two_gaps_of_one_batch_stay_two_rows() -> None: @@ -272,7 +298,11 @@ def test_two_gaps_of_one_batch_stay_two_rows() -> None: def lines_fail(acct, token): raise RuntimeError("a hop answered badly") - same_second = [invoice(payment_intent=None, total=3300), invoice(payment_intent=None, total=4400)] + # Distinct invoices carry distinct identities, which is what separates them. + same_second = [ + invoice(url=_hosted("entONE")[0], payment_intent=None, total=3300), + invoice(url=_hosted("entTWO")[0], payment_intent=None, total=4400), + ] keys = {unique_key_parts(r) for r in build_records(same_second, lines_fail)} assert len(keys) == 2, "two invoices sharing a second must not collapse into one key" diff --git a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py index e68d7f331..c59fc4628 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py +++ b/src/ingestion/connectors/ai/claude-team-invoices/tests/test_stream_over_http.py @@ -9,6 +9,7 @@ on an unregistered URL, so a hop that silently changes shape cannot pass. """ +import base64 import json import logging from collections.abc import Iterator, Mapping, Sequence @@ -27,7 +28,11 @@ "insight_source_id": "claude-team-invoices-1", } -ACCT, TOKEN = "acct_1EXAMPLE", "live_EXAMPLETOKEN" +ACCT = "acct_1EXAMPLE" +# base64 of `acct_1EXAMPLE,_entEXAMPLE,` — the shape the vendor emits, so +# the connector can decode an invoice identity out of it. +_PAYLOAD = f"{ACCT},_entEXAMPLE,1785456000-n1" +TOKEN = "live_" + base64.urlsafe_b64encode(_PAYLOAD.encode()).decode().rstrip("=") INVOICE_ID = "in_1EXAMPLE" EPHEMERAL_KEY = "ek_live_EXAMPLESECRET" diff --git a/src/ingestion/scripts/connectors-ddl/claude-team-invoices.sql b/src/ingestion/scripts/connectors-ddl/claude-team-invoices.sql index d877b6cdc..bb37c3b82 100644 --- a/src/ingestion/scripts/connectors-ddl/claude-team-invoices.sql +++ b/src/ingestion/scripts/connectors-ddl/claude-team-invoices.sql @@ -12,6 +12,7 @@ CREATE TABLE IF NOT EXISTS bronze_claude_team_invoices.claude_team_invoice_lines `data_source` Nullable(String), `collected_at` Nullable(String), `chain_status` Nullable(String), + `invoice_ref` Nullable(String), `invoice_id` Nullable(String), `invoice_status` Nullable(String), `invoice_created_ts` Nullable(Decimal(38, 9)), diff --git a/src/ingestion/tests/e2e/metrics/schemas/bronze_claude_team_invoices.claude_team_invoice_lines.yaml b/src/ingestion/tests/e2e/metrics/schemas/bronze_claude_team_invoices.claude_team_invoice_lines.yaml index 5e3db37b3..3c4f0b5d7 100644 --- a/src/ingestion/tests/e2e/metrics/schemas/bronze_claude_team_invoices.claude_team_invoice_lines.yaml +++ b/src/ingestion/tests/e2e/metrics/schemas/bronze_claude_team_invoices.claude_team_invoice_lines.yaml @@ -19,6 +19,7 @@ schemas: collected_at: { type: [string, "null"] } data_source: { type: [string, "null"] } chain_status: { type: [string, "null"] } + invoice_ref: { type: [string, "null"] } invoice_id: { type: [string, "null"] } invoice_status: { type: [string, "null"] } invoice_created_ts: { type: [number, "null"] } diff --git a/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py b/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py index 56ed49241..e59de4645 100644 --- a/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py +++ b/src/ingestion/tests/e2e/metrics/test_ai_invoice_silver.py @@ -76,6 +76,7 @@ def _base(source_id: str, read_at: str) -> dict: "collected_at": read_at, "data_source": "insight_claude_team", "chain_status": "ok", + "invoice_ref": None, "invoice_status": "paid", "invoice_created_ts": RAISED_AT, "invoice_currency": "usd", @@ -101,14 +102,29 @@ def _base(source_id: str, read_at: str) -> dict: } -def _invoice_row(intent: str, total: int, net: int, *, source_id: str = SOURCE, read_at: str = READ_AT, **over) -> dict: - """An invoice's own row, keyed on what the wrapper reports on every sync.""" +def _invoice_row( + intent: str, + total: int, + net: int, + *, + ref: str | None = None, + source_id: str = SOURCE, + read_at: str = READ_AT, + **over, +) -> dict: + """An invoice's own row. + + Keyed on the identity decoded out of its hosted URL when the vendor offered + one: that identity reads the same on every sync, which is what lets a later + sync replace this row instead of adding a second one beside it. With no URL + there is no identity, so the key falls back to what the wrapper reports — + both branches are exercised by the fixture below. + """ row = _base(source_id, read_at) + key = f"invoice-{ref}" if ref else f"invoice-{RAISED_AT}-{intent}-{total}-None" row.update( - # The key the connector builds when it has no invoice id to use, and keeps - # using once it has one — which is what lets a later sync replace this row. - # The chain outcome is deliberately absent from it. - unique_key=f"{TENANT}-{source_id}-invoice-{RAISED_AT}-{intent}-{total}-None", + unique_key=f"{TENANT}-{source_id}-{key}", + invoice_ref=ref, invoice_payment_intent=intent, invoice_total=total, invoice_total_excluding_tax=net, @@ -117,11 +133,18 @@ def _invoice_row(intent: str, total: int, net: int, *, source_id: str = SOURCE, return row -def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at: str = READ_AT, **over) -> dict: - """One line's row: line money only, keyed on Stripe's own ids.""" +def _line_row( + invoice_id: str, line_id: str, *, ref: str | None = None, source_id: str = SOURCE, read_at: str = READ_AT, **over +) -> dict: + """One line's row: line money only, keyed on Stripe's own ids. + + It carries its invoice's identity too, so a line can be tied back to the + invoice's row without going through an id the chain may not have reached. + """ row = _base(source_id, read_at) row.update( unique_key=f"{TENANT}-{source_id}-{invoice_id}-{line_id}", + invoice_ref=ref, invoice_id=invoice_id, line_id=line_id, category="subscriptions", @@ -142,6 +165,7 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at "pi_monthly", 16_500, 16_000, + ref="acct_EXAMPLE,_monthly", invoice_id="in_MONTHLY", invoice_num_seats=1, period_start_ts=AUG_START, @@ -150,6 +174,7 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at _line_row( "in_MONTHLY", "il_standard", + ref="acct_EXAMPLE,_monthly", tier_label="Standard", product_name="Example plan - Standard", description="1 x Example plan - Standard", @@ -161,6 +186,7 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at _line_row( "in_MONTHLY", "il_premium", + ref="acct_EXAMPLE,_monthly", tier_label="Premium", product_name="Example plan - Premium", description="5 x Example plan - Premium", @@ -171,11 +197,18 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at ), # A mid-period seat change: real money, no unit price. _invoice_row( - "pi_prorate", -1_500, -1_500, invoice_id="in_PRORATE", period_start_ts=AUG_START, period_end_ts=SEP_START + "pi_prorate", + -1_500, + -1_500, + ref="acct_EXAMPLE,_prorate", + invoice_id="in_PRORATE", + period_start_ts=AUG_START, + period_end_ts=SEP_START, ), _line_row( "in_PRORATE", "il_unused", + ref="acct_EXAMPLE,_prorate", description="Unused time on 5 x Example plan - Premium", amount=-1_500, quantity=5, @@ -183,11 +216,18 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at ), # Prepaid extra usage — the invoiced counterpart of used_credits. _invoice_row( - "pi_prepaid", 2_000, 2_000, invoice_id="in_PREPAID", period_start_ts=AUG_START, period_end_ts=SEP_START + "pi_prepaid", + 2_000, + 2_000, + ref="acct_EXAMPLE,_prepaid", + invoice_id="in_PREPAID", + period_start_ts=AUG_START, + period_end_ts=SEP_START, ), _line_row( "in_PREPAID", "il_prepaid", + ref="acct_EXAMPLE,_prepaid", category="overusage", description="Prepaid extra usage, Example plan", amount=2_000, @@ -196,7 +236,7 @@ def _line_row(invoice_id: str, line_id: str, *, source_id: str = SOURCE, read_at ), # An invoice whose chain never completed: the ledger survives, the price does # not, and with no line there is only the raise date to file it by. - _invoice_row("pi_broken", 999, 999, chain_status="failed", invoice_currency=None), + _invoice_row("pi_broken", 999, 999, ref="acct_EXAMPLE,_broken", chain_status="failed", invoice_currency=None), # A finalised invoice the vendor offered no hosted URL for. Not a draft: the # connector skips those, so one could never reach bronze in the first place. _invoice_row("pi_no_url", 750, 750, chain_status="no_hosted_url", invoice_status="open"), @@ -321,9 +361,16 @@ def test_an_invoice_with_no_line_falls_back_to_the_day_it_was_raised(invoice_sil # A recovery: the sync that failed and the sync that succeeded describe one # invoice, so the class has to end up with one row for it and not two. The pair # below is the same recovery twice — reached inside a single build, and reached -# across two, which is the case an append-only staging model gets wrong. +# across two, which is the case an append-only staging model gets wrong. Both +# syncs read the same identity out of the URL; that is what makes them one row. +RECOVERED_REF = "acct_EXAMPLE,_recovered" + + def _broken_row(source_id: str, read_at: str) -> dict: - return _invoice_row("pi_recovered", 16_500, 16_000, source_id=source_id, read_at=read_at, chain_status="failed") + """Carries the identity: a chain only fails once the URL has decoded.""" + return _invoice_row( + "pi_recovered", 16_500, 16_000, ref=RECOVERED_REF, source_id=source_id, read_at=read_at, chain_status="failed" + ) def _recovered_rows(source_id: str, read_at: str) -> list[dict]: @@ -332,6 +379,7 @@ def _recovered_rows(source_id: str, read_at: str) -> list[dict]: "pi_recovered", 16_500, 16_000, + ref=RECOVERED_REF, source_id=source_id, read_at=read_at, invoice_id="in_RECOVERED", @@ -341,6 +389,7 @@ def _recovered_rows(source_id: str, read_at: str) -> list[dict]: _line_row( "in_RECOVERED", "il_late", + ref=RECOVERED_REF, source_id=source_id, read_at=read_at, tier_label="Standard", @@ -407,6 +456,7 @@ def second_instance_silver( "pi_second", 4_200, 4_000, + ref="acct_EXAMPLE,_second", source_id=SOURCE_SECOND_INSTANCE, read_at=SECOND_INSTANCE_READ_AT, invoice_id="in_SECOND", From cae0a152c8574598acde2d6a8baeba35a02c6e96 Mon Sep 17 00:00:00 2001 From: Gregory Gogin Date: Wed, 19 Aug 2026 12:29:20 +0200 Subject: [PATCH 5/5] docs(ai-cost): say why the invoice staging model departs from the append convention MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The header said delete+insert was chosen because the same unique_key arrives twice. True, but it reads as a local preference, and the conventions check prescribes append for staging and delete+insert for silver — so the next reader to run that check sees a violation with no argument against it. The argument is the invariant: an invoice's row is REWRITTEN as its chain gets further, so a later sync has to replace the row an earlier one wrote rather than stand beside it. Appending leaves both until a background merge collapses them, which makes the replacement unobservable. The models the convention was written for restate a value under a key that never moves, which union_by_tag already resolves by _version — not this case. Comment only. Signed-off-by: Gregory Gogin --- .../dbt/claude_team__ai_invoice.sql | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql b/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql index 1db4f4009..71d9526ea 100644 --- a/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql +++ b/src/ingestion/connectors/ai/claude-team-invoices/dbt/claude_team__ai_invoice.sql @@ -24,10 +24,16 @@ -- 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. +-- STRATEGY: delete+insert, not append. A deliberate departure from the staging +-- convention (check-dbt-conventions prescribes delete+insert for silver and +-- append for staging), because this model carries an invariant the convention +-- does not anticipate: an invoice's row is REWRITTEN as its chain gets further, +-- so a later sync must replace the row an earlier one wrote rather than stand +-- beside it. Appending leaves both versions until a background merge collapses +-- them, which makes the replacement unobservable — the `unique` test reads +-- without FINAL. The models this convention was written for restate a value +-- under a key that never moves, which union_by_tag already resolves by +-- _version; that is not this case. See #2668. {{ config( materialized='incremental', incremental_strategy='delete+insert',