diff --git a/docs/analytics-v2-format.md b/docs/analytics-v2-format.md new file mode 100644 index 00000000..1f24ac3c --- /dev/null +++ b/docs/analytics-v2-format.md @@ -0,0 +1,60 @@ +# Completed-day analytics reports + +Public sharing is an explicit operator choice. Reports contain aggregate usage for completed UTC days. They do not contain prompts, customer identifiers, request identifiers, served upstream model identifiers, or private pricing diagnostics. + +A public-sharing activation begins coverage on the next full UTC day. Toggle days and collection-loss days are excluded. Private terminal records remain stored when public sharing is disabled. An omitted day means unavailable coverage; an included all-zero day means the collector covered the day and recorded no completed requests. + +## Signed event + +- Nostr kind: `38422`. +- Content schema: `routstr.analytics.v2`. +- Provider coordinate: `38421::`. +- Provider hash: the first 16 lowercase hexadecimal characters of SHA256 of the provider coordinate's UTF-8 bytes. +- `d` tag: `routstr.analytics.v2::week::epoch:`. +- `a` tag: the provider coordinate. +- `w` tag: the Monday date. + +The epoch distinguishes disjoint collection periods. Epochs can share a calendar week, but their covered dates must not overlap. A week is only the storage and transport grouping. Every reported day has its own totals and model partition, including elapsed days in the current week. + +Consumers verify the Nostr signature, provider coordinate, content, and date bounds before accepting a report. For each full signed coordinate, the greatest `created_at` wins, with the lexicographically lower event ID breaking ties. Corrections replace that coordinate's earlier values, not add to them. Conflicting overlapping epochs for one provider and day are unavailable coverage. + +## Content + +`week` is the Monday date. `epoch` is a nonnegative integer. `coverage_start` and `through` are inclusive dates within that week. `complete` means that the week or collection period is closed; it does not mean that every provider in the network reported. `corrected: true` identifies a corrected version, and `corrects` points to the immediately prior published version being corrected. An ordinary version can extend `through` but cannot alter earlier daily values or model partitions. + +`days` maps every date in the covered range to an integer vector. `daily_models` maps the same dates to objects mapping canonical model identifiers to vectors. Every day includes `_other`, and its named models plus `_other` sum exactly to the day's entire vector. Small-volume and free models can be named. When the signed frame would exceed its size limit, the smaller model rows are folded into `_other` without changing any daily total. Unknown models also contribute to `_other`. + +The `columns` array defines these positions, in order: + +| Position | Column | Meaning | +| --- | --- | --- | +| 0 | `completed_requests` | Successfully completed terminal settlements, including zero-charge completions | +| 1 | `input_observed_requests` | Requests with reported input usage | +| 2 | `output_observed_requests` | Requests with reported output usage | +| 3 | `cache_read_observed_requests` | Requests with reported cache-read usage | +| 4 | `cache_creation_observed_requests` | Requests with reported cache-creation usage | +| 5 | `input_tokens` | Reported or estimated input tokens | +| 6 | `output_tokens` | Reported or estimated output tokens | +| 7 | `cache_read_input_tokens` | Reported or estimated cache-read tokens | +| 8 | `cache_creation_input_tokens` | Reported or estimated cache-creation tokens | +| 9 | `revenue_msats` | Gross settled node revenue in millisatoshis | +| 10 | `input_estimated_requests` | Requests with estimated input usage | +| 11 | `output_estimated_requests` | Requests with estimated output usage | +| 12 | `cache_read_estimated_requests` | Requests with estimated cache-read usage | +| 13 | `cache_creation_estimated_requests` | Requests with estimated cache-creation usage | +| 14 | `input_missing_requests` | Requests without input usage | +| 15 | `output_missing_requests` | Requests without output usage | +| 16 | `cache_read_missing_requests` | Requests without cache-read usage | +| 17 | `cache_creation_missing_requests` | Requests without cache-creation usage | +| 18 | `measured_token_requests` | Requests eligible for the measured-token average | +| 19 | `measured_tokens` | Input, output, and cache tokens from exactly those eligible requests | + +For each input/output/cache component, reported + estimated + missing requests equals completed requests. Missing usage is not a measured zero. Every integer is nonnegative and no greater than JavaScript's maximum safe integer. Days and daily model rows must each satisfy these rules independently. + +The measured-token cohort requires reported input and output on the same request. Reported cache counts are included; missing cache counts of zero mean no recorded cache contribution. An estimated cache field, or a missing cache field with a positive count, excludes that request. Normalized input excludes cache tokens, so measured tokens add input, output, cache-read, and cache-creation exactly once. A reported zero-token completion remains eligible, including a free completion. + +The measured-token average is `measured_tokens / measured_token_requests`, or unavailable when the count is zero. Both fields are computed together from eligible requests. Separate input/output missing counters cannot reconstruct this cohort. Its request count cannot exceed either reported input or reported output request count; its tokens cannot exceed all four token totals; zero eligible requests require zero measured tokens. These conditions also hold for every model row and survive folding into `_other`. + +This unpublished format has exactly 20 columns. The earlier 18-column draft is not an accepted published format. Existing legacy public report formats remain separate inputs for consumers. + +The shared signed example is `tests/fixtures/analytics-v2/daily-models-signed.json`. It includes a paid model with reported usage and both cache components, a free model with estimated usage, an unknown model with missing usage, and actual zero-use days. Its three completions contain 190 recorded tokens and 3,250 msat of revenue. One completion qualifies for the measured average, with 170 tokens including 6 cache-read and 4 cache-creation tokens. diff --git a/routstr/nostr/analytics_v2.py b/routstr/nostr/analytics_v2.py new file mode 100644 index 00000000..c81dff5a --- /dev/null +++ b/routstr/nostr/analytics_v2.py @@ -0,0 +1,935 @@ +from __future__ import annotations + +import hashlib +import json +import unicodedata +from collections.abc import Iterable +from dataclasses import dataclass, replace +from datetime import date, timedelta +from typing import Any + +from nostr_sdk import ( + Event, + EventBuilder, + Keys, + Kind, + Tag, + Timestamp, +) + +ANALYTICS_KIND = 38422 +ANALYTICS_SCHEMA = "routstr.analytics.v2" +DEFAULT_MAX_FRAME_BYTES = 96 * 1024 +MAX_SAFE_INTEGER = (1 << 53) - 1 + +COLUMNS = ( + "completed_requests", + "input_observed_requests", + "output_observed_requests", + "cache_read_observed_requests", + "cache_creation_observed_requests", + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "revenue_msats", + "input_estimated_requests", + "output_estimated_requests", + "cache_read_estimated_requests", + "cache_creation_estimated_requests", + "input_missing_requests", + "output_missing_requests", + "cache_read_missing_requests", + "cache_creation_missing_requests", + "measured_token_requests", + "measured_tokens", +) + +MetricVector = tuple[int, ...] +ZERO_VECTOR: MetricVector = (0,) * len(COLUMNS) + + +class AnalyticsV2Error(ValueError): + pass + + +class FrameTooLargeError(AnalyticsV2Error): + def __init__(self, frame_size: int, max_frame_bytes: int) -> None: + self.frame_size = frame_size + self.max_frame_bytes = max_frame_bytes + super().__init__( + f"Analytics v2 frame is {frame_size} bytes, limit is {max_frame_bytes}" + ) + + +@dataclass(frozen=True) +class LedgerOutcome: + terminal_day: date + model_identifier: str | None + input_tokens: int + output_tokens: int + cache_read_input_tokens: int + cache_creation_input_tokens: int + revenue_msats: int + input_source: str + output_source: str + cache_read_source: str + cache_creation_source: str + completed_requests: int = 1 + + +@dataclass(frozen=True) +class DayAggregate: + day: date + values: MetricVector + + +@dataclass(frozen=True) +class ModelAggregate: + identifier: str | None + values: MetricVector + + +@dataclass(frozen=True) +class DailyModelAggregate: + day: date + models: tuple[ModelAggregate, ...] + + +@dataclass(frozen=True) +class PriorVersion: + event_id: str + pubkey: str + d_tag: str + created_at: int + week: date + epoch: int + coverage_start: date + through: date + complete: bool + corrected: bool + days: tuple[DayAggregate, ...] + daily_models: tuple[DailyModelAggregate, ...] + + +@dataclass(frozen=True) +class WeeklyAggregate: + week: date + epoch: int + coverage_start: date + through: date + complete: bool + days: tuple[DayAggregate, ...] + daily_models: tuple[DailyModelAggregate, ...] + prior_version: PriorVersion | None = None + correction: bool = False + + +@dataclass(frozen=True) +class AnalyticsAddress: + coordinate: str + provider_hash: str + d_tag: str + week: date + tags: tuple[tuple[str, str], ...] + + +@dataclass(frozen=True) +class EncodedAnalyticsEvent: + event_id: str + pubkey: str + created_at: int + signature: str + tags: tuple[tuple[str, str], ...] + content: bytes + frame: bytes + d_tag: str + week: date + epoch: int + coverage_start: date + through: date + complete: bool + corrected: bool + days: tuple[DayAggregate, ...] + daily_models: tuple[DailyModelAggregate, ...] + + @property + def event(self) -> dict[str, Any]: + return _event_dict( + self.event_id, + self.pubkey, + self.created_at, + self.tags, + self.content.decode("utf-8"), + self.signature, + ) + + def as_prior_version(self) -> PriorVersion: + return PriorVersion( + event_id=self.event_id, + pubkey=self.pubkey, + d_tag=self.d_tag, + created_at=self.created_at, + week=self.week, + epoch=self.epoch, + coverage_start=self.coverage_start, + through=self.through, + complete=self.complete, + corrected=self.corrected, + days=self.days, + daily_models=self.daily_models, + ) + + +def build_analytics_address( + public_key_hex: str, provider_d: str, week: date, epoch: int = 0 +) -> AnalyticsAddress: + _validate_public_key(public_key_hex) + _validate_provider_d(provider_d) + _validate_week(week) + _validate_epoch(epoch) + coordinate = f"38421:{public_key_hex}:{provider_d}" + provider_hash = hashlib.sha256(coordinate.encode("utf-8")).hexdigest()[:16] + d_tag = ( + f"routstr.analytics.v2:{provider_hash}:week:{week.isoformat()}:epoch:{epoch}" + ) + return AnalyticsAddress( + coordinate=coordinate, + provider_hash=provider_hash, + d_tag=d_tag, + week=week, + tags=( + ("d", d_tag), + ("a", coordinate), + ("w", week.isoformat()), + ), + ) + + +def aggregate_ledger_week( + outcomes: Iterable[LedgerOutcome], + *, + epoch: int, + epoch_coverage_start: date, + epoch_coverage_end: date | None, + week: date, + today_utc: date, + prior_version: PriorVersion | None = None, + correction: bool = False, +) -> WeeklyAggregate | None: + """Aggregate the covered part of one week, capped at yesterday in UTC.""" + _validate_epoch(epoch) + _validate_date(epoch_coverage_start, "epoch_coverage_start") + if epoch_coverage_end is not None: + _validate_date(epoch_coverage_end, "epoch_coverage_end") + _validate_week(week) + _validate_date(today_utc, "today_utc") + if not isinstance(correction, bool): + raise AnalyticsV2Error("correction must be a boolean") + + coverage_start = max(epoch_coverage_start, week) + last_complete_day = today_utc - timedelta(days=1) + week_end = week + timedelta(days=6) + through = min(last_complete_day, week_end) + if epoch_coverage_end is not None: + through = min(through, epoch_coverage_end) + if through < coverage_start: + return None + + daily = {day: ZERO_VECTOR for day in _date_range(coverage_start, through)} + models: dict[date, dict[str | None, MetricVector]] = {day: {} for day in daily} + for outcome in outcomes: + _validate_date(outcome.terminal_day, "terminal_day") + if not (coverage_start <= outcome.terminal_day <= through): + continue + _validate_outcome(outcome) + values = _outcome_vector(outcome) + daily[outcome.terminal_day] = _add_vectors(daily[outcome.terminal_day], values) + identifier = ( + outcome.model_identifier + if _valid_model_identifier(outcome.model_identifier) + else None + ) + day_models = models[outcome.terminal_day] + day_models[identifier] = _add_vectors( + day_models.get(identifier, ZERO_VECTOR), values + ) + + model_rows = tuple( + DailyModelAggregate( + day, + tuple( + ModelAggregate(identifier, values) + for identifier, values in sorted( + rows.items(), key=lambda item: (item[0] is None, item[0] or "") + ) + ), + ) + for day, rows in models.items() + ) + return WeeklyAggregate( + week=week, + epoch=epoch, + coverage_start=coverage_start, + through=through, + complete=through == min(week_end, epoch_coverage_end or week_end), + days=tuple(DayAggregate(day, values) for day, values in daily.items()), + daily_models=model_rows, + prior_version=prior_version, + correction=correction, + ) + + +def encode_week_event( + aggregate: WeeklyAggregate, + *, + private_key_hex: str, + provider_d: str, + created_at: int, + max_frame_bytes: int = DEFAULT_MAX_FRAME_BYTES, +) -> EncodedAnalyticsEvent: + """Build, sign and size-fold one immutable analytics v2 EVENT frame.""" + _validate_aggregate(aggregate) + _validate_nonnegative_int(created_at, "created_at", maximum=None) + _validate_positive_int(max_frame_bytes, "max_frame_bytes") + private_key = _private_key(private_key_hex) + public_key = private_key.public_key().to_hex() + address = build_analytics_address( + public_key, provider_d, aggregate.week, aggregate.epoch + ) + corrected, corrects = _validate_successor( + aggregate, address, created_at, public_key + ) + base = _base_payload(aggregate, corrected=corrected, corrects=corrects) + maximum = max((len(row.models) for row in aggregate.daily_models), default=0) + best: EncodedAnalyticsEvent | None = None + lower, upper = 0, maximum + while lower <= upper: + kept_count = (lower + upper) // 2 + payload = dict(base) + payload["daily_models"] = _wire_daily_models(aggregate, kept_count) + encoded = _sign_payload( + payload, aggregate, address, private_key, public_key, created_at, corrected + ) + if len(encoded.frame) <= max_frame_bytes: + best = encoded + lower = kept_count + 1 + else: + upper = kept_count - 1 + if best is None: + raise FrameTooLargeError(len(encoded.frame), max_frame_bytes) + prior = aggregate.prior_version + if prior is not None and not aggregate.correction: + previous = {row.day: row.models for row in prior.daily_models} + current = {row.day: row.models for row in best.daily_models} + if any(current.get(day) != models for day, models in previous.items()): + return encode_week_event( + replace(aggregate, correction=True), + private_key_hex=private_key_hex, + provider_d=provider_d, + created_at=created_at, + max_frame_bytes=max_frame_bytes, + ) + return best + + +def prior_version_from_frame(frame: bytes) -> PriorVersion: + """Verify an encoder-owned EVENT frame and recover its successor state.""" + if not isinstance(frame, bytes): + raise AnalyticsV2Error("frame must be bytes") + try: + message = json.loads(frame.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise AnalyticsV2Error("Invalid analytics EVENT frame") from error + if ( + not isinstance(message, list) + or len(message) != 2 + or message[0] != "EVENT" + or not isinstance(message[1], dict) + ): + raise AnalyticsV2Error("Invalid analytics EVENT frame") + + event = message[1] + required_event_keys = { + "id", + "pubkey", + "created_at", + "kind", + "tags", + "content", + "sig", + } + if set(event) != required_event_keys: + raise AnalyticsV2Error("Invalid analytics event fields") + event_id = event["id"] + public_key = event["pubkey"] + created_at = event["created_at"] + signature = event["sig"] + content_text = event["content"] + _validate_hex(event_id, 64, "event id") + _validate_public_key(public_key) + _validate_nonnegative_int(created_at, "created_at", maximum=None) + _validate_hex(signature, 128, "signature") + if event["kind"] != ANALYTICS_KIND or isinstance(event["kind"], bool): + raise AnalyticsV2Error("Invalid analytics event kind") + if not isinstance(content_text, str): + raise AnalyticsV2Error("Analytics event content must be a string") + + tags = _parse_wire_tags(event["tags"]) + canonical_event = _event_dict( + event_id, public_key, created_at, tags, content_text, signature + ) + if frame != _frame_bytes(canonical_event): + raise AnalyticsV2Error("Analytics EVENT frame is not canonical") + + try: + signature_valid = Event.from_json(json.dumps(event)).verify() + except Exception as error: + raise AnalyticsV2Error("Invalid analytics event signature") from error + if not signature_valid: + raise AnalyticsV2Error("Invalid analytics event signature") + + try: + payload = json.loads(content_text) + except json.JSONDecodeError as error: + raise AnalyticsV2Error("Invalid analytics content JSON") from error + if not isinstance(payload, dict): + raise AnalyticsV2Error("Analytics content must be an object") + if _canonical_json(payload) != content_text.encode("utf-8"): + raise AnalyticsV2Error("Analytics content is not canonical") + if payload.get("schema") != ANALYTICS_SCHEMA: + raise AnalyticsV2Error("Invalid analytics schema") + if payload.get("columns") != list(COLUMNS): + raise AnalyticsV2Error("Invalid analytics columns") + + week = _parse_wire_date(payload.get("week"), "week") + _validate_week(week) + coverage_start = _parse_wire_date(payload.get("coverage_start"), "coverage_start") + through = _parse_wire_date(payload.get("through"), "through") + epoch_value = payload.get("epoch") + if isinstance(epoch_value, bool) or not isinstance(epoch_value, int): + raise AnalyticsV2Error("epoch must be a non-negative integer") + epoch = epoch_value + complete = payload.get("complete") + _validate_epoch(epoch) + if not isinstance(complete, bool): + raise AnalyticsV2Error("complete must be a boolean") + + corrected_value = payload.get("corrected", False) + if corrected_value is not False and corrected_value is not True: + raise AnalyticsV2Error("corrected must be a boolean") + if "corrected" in payload and corrected_value is not True: + raise AnalyticsV2Error("corrected may only be present when true") + corrected = corrected_value is True + if "corrects" in payload: + _validate_hex(payload["corrects"], 64, "corrects") + if not corrected: + raise AnalyticsV2Error("corrects requires corrected true") + + days = _parse_wire_days(payload.get("days"), coverage_start, through) + daily_models = _parse_wire_daily_models(payload.get("daily_models"), days) + aggregate = WeeklyAggregate( + week=week, + epoch=epoch, + coverage_start=coverage_start, + through=through, + complete=complete, + days=days, + daily_models=daily_models, + ) + _validate_aggregate(aggregate) + + if tags[2] != ("w", week.isoformat()): + raise AnalyticsV2Error("Week tag does not match content") + coordinate_parts = tags[1][1].split(":", 2) + if len(coordinate_parts) != 3 or coordinate_parts[:2] != ["38421", public_key]: + raise AnalyticsV2Error("Invalid analytics provider coordinate") + address = build_analytics_address(public_key, coordinate_parts[2], week, epoch) + if tags != address.tags: + raise AnalyticsV2Error("Analytics tags do not match the signed coordinate") + + return PriorVersion( + event_id=event_id, + pubkey=public_key, + d_tag=address.d_tag, + created_at=created_at, + week=week, + epoch=epoch, + coverage_start=coverage_start, + through=through, + complete=complete, + corrected=corrected, + days=days, + daily_models=daily_models, + ) + + +def _base_payload( + aggregate: WeeklyAggregate, *, corrected: bool, corrects: str | None +) -> dict[str, Any]: + payload: dict[str, Any] = { + "schema": ANALYTICS_SCHEMA, + "week": aggregate.week.isoformat(), + "epoch": aggregate.epoch, + "coverage_start": aggregate.coverage_start.isoformat(), + "through": aggregate.through.isoformat(), + "complete": aggregate.complete, + "columns": list(COLUMNS), + "days": {row.day.isoformat(): list(row.values) for row in aggregate.days}, + } + if corrects is not None: + payload["corrects"] = corrects + if corrected: + payload["corrected"] = True + return payload + + +def _canonical_json(value: object) -> bytes: + return json.dumps(value, separators=(",", ":"), sort_keys=True).encode("utf-8") + + +def _event_dict( + event_id: str, + public_key: str, + created_at: int, + tags: tuple[tuple[str, str], ...], + content: str, + signature: str, +) -> dict[str, Any]: + return { + "id": event_id, + "pubkey": public_key, + "created_at": created_at, + "kind": ANALYTICS_KIND, + "tags": [list(tag) for tag in tags], + "content": content, + "sig": signature, + } + + +def _frame_bytes(event: dict[str, Any]) -> bytes: + return json.dumps( + ["EVENT", event], separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + + +def _parse_wire_tags(value: object) -> tuple[tuple[str, str], ...]: + if not isinstance(value, list) or len(value) != 3: + raise AnalyticsV2Error("Analytics event must carry exactly three tags") + tags: list[tuple[str, str]] = [] + for tag in value: + if ( + not isinstance(tag, list) + or len(tag) != 2 + or not all(isinstance(item, str) for item in tag) + ): + raise AnalyticsV2Error("Invalid analytics event tag") + tags.append((tag[0], tag[1])) + if tuple(tag[0] for tag in tags) != ("d", "a", "w"): + raise AnalyticsV2Error("Analytics tags must be d, a and w") + return tuple(tags) + + +def _parse_wire_date(value: object, name: str) -> date: + if not isinstance(value, str): + raise AnalyticsV2Error(f"{name} must be an ISO date") + try: + parsed = date.fromisoformat(value) + except ValueError as error: + raise AnalyticsV2Error(f"{name} must be an ISO date") from error + if parsed.isoformat() != value: + raise AnalyticsV2Error(f"{name} must be an ISO date") + return parsed + + +def _parse_wire_days( + value: object, coverage_start: date, through: date +) -> tuple[DayAggregate, ...]: + if not isinstance(value, dict): + raise AnalyticsV2Error("days must be an object") + expected_days = _date_range(coverage_start, through) + expected_keys = {day.isoformat() for day in expected_days} + if set(value) != expected_keys: + raise AnalyticsV2Error("days do not match the covered range") + rows: list[DayAggregate] = [] + for day in expected_days: + raw_values = value[day.isoformat()] + if not isinstance(raw_values, list): + raise AnalyticsV2Error("Daily metrics must be an array") + values = tuple(raw_values) + _validate_vector(values) + rows.append(DayAggregate(day, values)) + return tuple(rows) + + +def _parse_wire_daily_models( + value: object, days: tuple[DayAggregate, ...] +) -> tuple[DailyModelAggregate, ...]: + if not isinstance(value, dict) or set(value) != { + row.day.isoformat() for row in days + }: + raise AnalyticsV2Error("daily_models must exactly cover the days") + rows: list[DailyModelAggregate] = [] + for day in days: + models = value[day.day.isoformat()] + if not isinstance(models, dict) or "_other" not in models: + raise AnalyticsV2Error("Daily model rows require _other") + model_rows: list[ModelAggregate] = [] + for identifier, raw_values in sorted(models.items()): + if ( + not isinstance(identifier, str) + or not identifier + or not isinstance(raw_values, list) + ): + raise AnalyticsV2Error("Invalid daily model row") + values = tuple(raw_values) + _validate_vector(values) + model_rows.append(ModelAggregate(identifier, values)) + rows.append(DailyModelAggregate(day.day, tuple(model_rows))) + return tuple(rows) + + +def _sign_payload( + payload: dict[str, Any], + aggregate: WeeklyAggregate, + address: AnalyticsAddress, + private_key: Keys, + public_key: str, + created_at: int, + corrected: bool, +) -> EncodedAnalyticsEvent: + content = _canonical_json(payload) + content_text = content.decode("utf-8") + signed = ( + EventBuilder(Kind(ANALYTICS_KIND), content_text) + .tags([Tag.parse(list(tag)) for tag in address.tags]) + .custom_created_at(Timestamp.from_secs(created_at)) + .finalize(private_key) + ) + event = json.loads(signed.as_json()) + event_id = event["id"] + signature = event["sig"] + return EncodedAnalyticsEvent( + event_id=event_id, + pubkey=public_key, + created_at=created_at, + signature=signature, + tags=address.tags, + content=content, + frame=_frame_bytes(event), + d_tag=address.d_tag, + week=aggregate.week, + epoch=aggregate.epoch, + coverage_start=aggregate.coverage_start, + through=aggregate.through, + complete=aggregate.complete, + corrected=corrected, + days=aggregate.days, + daily_models=_parse_wire_daily_models(payload["daily_models"], aggregate.days), + ) + + +def _validate_successor( + aggregate: WeeklyAggregate, + address: AnalyticsAddress, + created_at: int, + public_key: str, +) -> tuple[bool, str | None]: + prior = aggregate.prior_version + if prior is None: + if aggregate.correction: + raise AnalyticsV2Error( + "A correction requires the immediately prior version" + ) + return False, None + + _validate_prior(prior) + if prior.pubkey != public_key or prior.d_tag != address.d_tag: + raise AnalyticsV2Error("Prior version belongs to a different coordinate") + if prior.week != aggregate.week: + raise AnalyticsV2Error("Prior version belongs to a different week") + if created_at <= prior.created_at: + raise AnalyticsV2Error("created_at must strictly increase") + if aggregate.epoch < prior.epoch: + raise AnalyticsV2Error("epoch must not decrease") + + current_days = {row.day: row.values for row in aggregate.days} + prior_days = {row.day: row.values for row in prior.days} + if aggregate.epoch == prior.epoch: + if aggregate.coverage_start != prior.coverage_start: + raise AnalyticsV2Error("coverage_start changed within an epoch") + if aggregate.through < prior.through: + raise AnalyticsV2Error("through must not decrease") + if prior.complete and not aggregate.complete: + raise AnalyticsV2Error("complete must not revert") + if aggregate.correction: + return True, prior.event_id + if prior.complete: + raise AnalyticsV2Error("A closed week can only be corrected") + if aggregate.through == prior.through: + raise AnalyticsV2Error("An ordinary version must advance through") + if any( + current_days.get(day) != values for day, values in prior_days.items() + ) or daily_models_changed(aggregate, prior): + raise AnalyticsV2Error("An ordinary version changed a published row") + else: + if aggregate.correction: + raise AnalyticsV2Error("A correction cannot change epoch") + if current_days.keys() & prior_days.keys(): + raise AnalyticsV2Error("A new epoch must not copy prior rows") + + return prior.corrected, None + + +def _validate_aggregate(aggregate: WeeklyAggregate) -> None: + _validate_period( + aggregate.week, + aggregate.epoch, + aggregate.coverage_start, + aggregate.through, + aggregate.complete, + aggregate.days, + ) + if not isinstance(aggregate.correction, bool): + raise AnalyticsV2Error("correction must be a boolean") + if tuple(row.day for row in aggregate.daily_models) != tuple( + row.day for row in aggregate.days + ): + raise AnalyticsV2Error("daily_models must exactly cover the days") + for day, models in zip(aggregate.days, aggregate.daily_models): + identifiers: set[str | None] = set() + model_total = ZERO_VECTOR + for model in models.models: + if model.identifier in identifiers: + raise AnalyticsV2Error("model identifiers must be unique within a day") + identifiers.add(model.identifier) + _validate_vector(model.values) + model_total = _add_vectors(model_total, model.values) + if model_total != day.values: + raise AnalyticsV2Error("daily model rows must partition that day's totals") + + +def _validate_prior(prior: PriorVersion) -> None: + _validate_hex(prior.event_id, 64, "prior event id") + _validate_public_key(prior.pubkey) + _validate_nonnegative_int(prior.created_at, "prior created_at", maximum=None) + _validate_period( + prior.week, + prior.epoch, + prior.coverage_start, + prior.through, + prior.complete, + prior.days, + ) + if not isinstance(prior.corrected, bool): + raise AnalyticsV2Error("prior corrected must be a boolean") + + +def _validate_period( + week: date, + epoch: int, + coverage_start: date, + through: date, + complete: bool, + days: tuple[DayAggregate, ...], +) -> None: + _validate_week(week) + _validate_epoch(epoch) + _validate_date(coverage_start, "coverage_start") + _validate_date(through, "through") + if coverage_start < week or coverage_start > through: + raise AnalyticsV2Error("Invalid coverage range") + if through > week + timedelta(days=6): + raise AnalyticsV2Error("through follows the event week") + if not isinstance(complete, bool): + raise AnalyticsV2Error("complete must be a boolean") + if through == week + timedelta(days=6) and not complete: + raise AnalyticsV2Error("A week covering Sunday must be complete") + if tuple(row.day for row in days) != _date_range(coverage_start, through): + raise AnalyticsV2Error("days must exactly cover the coverage range") + for row in days: + _validate_vector(row.values) + + +def daily_models_changed(aggregate: WeeklyAggregate, prior: PriorVersion) -> bool: + # Compare the same public detail level after size folding. + prior_limit = max( + ( + sum(row.identifier != "_other" for row in day.models) + for day in prior.daily_models + ), + default=0, + ) + current = _wire_daily_models(aggregate, prior_limit) + return any( + current.get(day.day.isoformat()) + != {row.identifier: list(row.values) for row in day.models} + for day in prior.daily_models + ) + + +def _wire_daily_models( + aggregate: WeeklyAggregate, keep_count: int +) -> dict[str, dict[str, list[int]]]: + result: dict[str, dict[str, list[int]]] = {} + for day, models in zip(aggregate.days, aggregate.daily_models): + named = sorted( + (row for row in models.models if _valid_model_identifier(row.identifier)), + key=lambda row: (-row.values[0], row.identifier or ""), + )[:keep_count] + named_total = _sum_vectors(row.values for row in named) + other = tuple(total - value for total, value in zip(day.values, named_total)) + result[day.day.isoformat()] = { + row.identifier: list(row.values) + for row in named + if row.identifier is not None + } + result[day.day.isoformat()]["_other"] = list(other) + return result + + +def _outcome_vector(outcome: LedgerOutcome) -> MetricVector: + sources = ( + outcome.input_source, + outcome.output_source, + outcome.cache_read_source, + outcome.cache_creation_source, + ) + tokens = ( + outcome.input_tokens, + outcome.output_tokens, + outcome.cache_read_input_tokens, + outcome.cache_creation_input_tokens, + ) + measured = sources[:2] == ("reported", "reported") and all( + source == "reported" or (source == "missing" and value == 0) + for source, value in zip(sources[2:], tokens[2:]) + ) + return ( + outcome.completed_requests, + *(outcome.completed_requests * int(source == "reported") for source in sources), + *tokens, + outcome.revenue_msats, + *( + outcome.completed_requests * int(source == "estimated") + for source in sources + ), + *(outcome.completed_requests * int(source == "missing") for source in sources), + outcome.completed_requests if measured else 0, + sum(tokens) if measured else 0, + ) + + +def _validate_outcome(outcome: LedgerOutcome) -> None: + _validate_date(outcome.terminal_day, "terminal_day") + _validate_positive_int(outcome.completed_requests, "completed_requests") + for name in ("input", "output", "cache_read", "cache_creation"): + source = getattr(outcome, f"{name}_source") + if source not in ("reported", "estimated", "missing"): + raise AnalyticsV2Error("Invalid usage provenance") + for name in ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", + "revenue_msats", + ): + _validate_nonnegative_int(getattr(outcome, name), name) + + +def _validate_vector(values: MetricVector) -> None: + if not isinstance(values, tuple) or len(values) != len(COLUMNS): + raise AnalyticsV2Error( + f"Metric vectors must contain exactly {len(COLUMNS)} integers" + ) + for index, value in enumerate(values): + _validate_nonnegative_int(value, COLUMNS[index]) + for index in range(4): + if values[1 + index] + values[10 + index] + values[14 + index] != values[0]: + raise AnalyticsV2Error("Usage provenance must partition completed requests") + if values[18] > min(values[1], values[2]): + raise AnalyticsV2Error("Measured requests require reported input and output") + if values[19] > sum(values[5:9]) or (values[18] == 0 and values[19] != 0): + raise AnalyticsV2Error("Measured tokens must belong to measured requests") + + +def _sum_vectors(vectors: Iterable[MetricVector]) -> MetricVector: + total = ZERO_VECTOR + for values in vectors: + total = _add_vectors(total, values) + return total + + +def _add_vectors(left: MetricVector, right: MetricVector) -> MetricVector: + result = tuple(a + b for a, b in zip(left, right)) + if any(value > MAX_SAFE_INTEGER for value in result): + raise AnalyticsV2Error("An aggregate exceeds the maximum safe JSON integer") + return result + + +def _date_range(start: date, end: date) -> tuple[date, ...]: + return tuple( + start + timedelta(days=offset) for offset in range((end - start).days + 1) + ) + + +def _valid_model_identifier(value: object) -> bool: + return isinstance(value, str) and bool(value) and value != "_other" + + +def _private_key(private_key_hex: str) -> Keys: + _validate_hex(private_key_hex, 64, "private key") + try: + return Keys.parse(private_key_hex) + except Exception as error: + raise AnalyticsV2Error("Invalid private key") from error + + +def _validate_public_key(value: str) -> None: + _validate_hex(value, 64, "public key") + + +def _validate_provider_d(value: str) -> None: + if not isinstance(value, str) or not 1 <= len(value) <= 64: + raise AnalyticsV2Error("provider_d must contain 1 to 64 characters") + if any(unicodedata.category(character) == "Cc" for character in value): + raise AnalyticsV2Error("provider_d must not contain control characters") + + +def _validate_week(value: date) -> None: + _validate_date(value, "week") + if value.weekday() != 0: + raise AnalyticsV2Error("week must be a Monday") + + +def _validate_epoch(value: int) -> None: + _validate_nonnegative_int(value, "epoch", maximum=None) + + +def _validate_date(value: object, name: str) -> None: + if type(value) is not date: + raise AnalyticsV2Error(f"{name} must be a date") + + +def _validate_positive_int(value: object, name: str) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise AnalyticsV2Error(f"{name} must be a positive integer") + + +def _validate_nonnegative_int( + value: object, name: str, *, maximum: int | None = MAX_SAFE_INTEGER +) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise AnalyticsV2Error(f"{name} must be a non-negative integer") + if maximum is not None and value > maximum: + raise AnalyticsV2Error(f"{name} exceeds the maximum safe JSON integer") + + +def _validate_hex(value: object, length: int, name: str) -> None: + if ( + not isinstance(value, str) + or len(value) != length + or value != value.lower() + or any(character not in "0123456789abcdef" for character in value) + ): + raise AnalyticsV2Error(f"{name} must be {length} lowercase hex characters") diff --git a/tests/fixtures/analytics-v2/daily-models-signed.json b/tests/fixtures/analytics-v2/daily-models-signed.json new file mode 100644 index 00000000..b5508c0b --- /dev/null +++ b/tests/fixtures/analytics-v2/daily-models-signed.json @@ -0,0 +1,22 @@ +{ + "id": "46c2824bbfd86045982f291c725fb0c91c27b3dc7ce43a2f3b70f16b3fda5c3c", + "pubkey": "4f355bdcb7cc0af728ef3cceb9615d90684bb5b2ca5f859ab0f0b704075871aa", + "created_at": 1789689600, + "kind": 38422, + "tags": [ + [ + "d", + "routstr.analytics.v2:699b4243f30e4058:week:2026-09-14:epoch:3" + ], + [ + "a", + "38421:4f355bdcb7cc0af728ef3cceb9615d90684bb5b2ca5f859ab0f0b704075871aa:provider" + ], + [ + "w", + "2026-09-14" + ] + ], + "content": "{\"columns\":[\"completed_requests\",\"input_observed_requests\",\"output_observed_requests\",\"cache_read_observed_requests\",\"cache_creation_observed_requests\",\"input_tokens\",\"output_tokens\",\"cache_read_input_tokens\",\"cache_creation_input_tokens\",\"revenue_msats\",\"input_estimated_requests\",\"output_estimated_requests\",\"cache_read_estimated_requests\",\"cache_creation_estimated_requests\",\"input_missing_requests\",\"output_missing_requests\",\"cache_read_missing_requests\",\"cache_creation_missing_requests\",\"measured_token_requests\",\"measured_tokens\"],\"complete\":false,\"coverage_start\":\"2026-09-14\",\"daily_models\":{\"2026-09-14\":{\"_other\":[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],\"model/free\":[1,0,0,0,0,15,5,0,0,0,1,1,0,0,0,0,1,1,0,0],\"openai/gpt-test\":[1,1,1,1,1,120,40,6,4,2500,0,0,0,0,0,0,0,0,1,170]},\"2026-09-15\":{\"_other\":[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]},\"2026-09-16\":{\"_other\":[1,0,0,0,0,0,0,0,0,750,0,0,0,0,1,1,1,1,0,0]},\"2026-09-17\":{\"_other\":[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]}},\"days\":{\"2026-09-14\":[2,1,1,1,1,135,45,6,4,2500,1,1,0,0,0,0,1,1,1,170],\"2026-09-15\":[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],\"2026-09-16\":[1,0,0,0,0,0,0,0,0,750,0,0,0,0,1,1,1,1,0,0],\"2026-09-17\":[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]},\"epoch\":3,\"schema\":\"routstr.analytics.v2\",\"through\":\"2026-09-17\",\"week\":\"2026-09-14\"}", + "sig": "f03350c4bcf6f7ab7b455352f95ccbb2a9a65215e291deacea9291b3c79972f8d8e59255836cde23491c2443a695e508b163b9f3c41d21e23b6f7b988b716638" +} diff --git a/tests/unit/test_nostr_analytics_v2.py b/tests/unit/test_nostr_analytics_v2.py new file mode 100644 index 00000000..5552f23f --- /dev/null +++ b/tests/unit/test_nostr_analytics_v2.py @@ -0,0 +1,529 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import replace +from datetime import date, timedelta +from pathlib import Path +from typing import Any + +import pytest +from nostr_sdk import Event, EventBuilder, Keys, Kind, Tag, Timestamp + +from routstr.nostr.analytics_v2 import ( + COLUMNS, + DEFAULT_MAX_FRAME_BYTES, + MAX_SAFE_INTEGER, + AnalyticsV2Error, + DayAggregate, + FrameTooLargeError, + LedgerOutcome, + aggregate_ledger_week, + build_analytics_address, + encode_week_event, + prior_version_from_frame, +) + +PRIVATE_KEY = "11" * 32 +WEEK = date(2026, 9, 14) + + +def _outcome( + day: date = WEEK, model: str | None = "model/a", **changes: Any +) -> LedgerOutcome: + return replace( + LedgerOutcome( + terminal_day=day, + model_identifier=model, + input_tokens=10, + output_tokens=4, + cache_read_input_tokens=0, + cache_creation_input_tokens=0, + revenue_msats=25, + input_source="reported", + output_source="estimated", + cache_read_source="missing", + cache_creation_source="missing", + ), + **changes, + ) + + +def _aggregate(outcomes: list[LedgerOutcome] | None = None, **changes: Any) -> Any: + args: dict[str, Any] = dict( + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK + timedelta(days=4), + ) + args.update(changes) + aggregate = aggregate_ledger_week(outcomes or [], **args) + assert aggregate is not None + return aggregate + + +def _encode(aggregate: Any, **changes: Any) -> Any: + args: dict[str, Any] = dict( + private_key_hex=PRIVATE_KEY, provider_d="provider", created_at=1789689600 + ) + args.update(changes) + return encode_week_event(aggregate, **args) + + +def _payload(encoded: Any) -> dict[str, Any]: + return json.loads(encoded.content) + + +def _resign(encoded: Any, payload: dict[str, Any]) -> bytes: + event = ( + EventBuilder( + Kind(38422), + json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ), + ) + .tags([Tag.parse(list(tag)) for tag in encoded.tags]) + .custom_created_at(Timestamp.from_secs(encoded.created_at)) + .finalize(Keys.parse(PRIVATE_KEY)) + ) + return json.dumps( + ["EVENT", json.loads(event.as_json())], + separators=(",", ":"), + ensure_ascii=False, + ).encode() + + +def _assert_partitions(payload: dict[str, Any]) -> None: + assert payload["daily_models"].keys() == payload["days"].keys() + for day, values in payload["days"].items(): + models = payload["daily_models"][day] + assert "_other" in models + assert [ + sum(row[column] for row in models.values()) + for column in range(len(COLUMNS)) + ] == values + for row in [values, *models.values()]: + for side in range(4): + assert row[1 + side] + row[10 + side] + row[14 + side] == row[0] + assert row[18] <= min(row[1], row[2]) + assert row[19] <= sum(row[5:9]) + assert row[18] != 0 or row[19] == 0 + + +def test_partial_week_has_daily_models_and_zero_days() -> None: + rows = [ + _outcome(), + _outcome(WEEK, "model/free", revenue_msats=0), + _outcome(WEEK + timedelta(days=2), None), + _outcome(WEEK + timedelta(days=4), "today/not-published"), + _outcome(WEEK - timedelta(days=1), "before/not-published"), + ] + encoded = _encode(_aggregate(rows)) + payload = _payload(encoded) + assert payload["complete"] is False + assert payload["through"] == "2026-09-17" + assert payload["days"]["2026-09-14"][0] == 2 + assert payload["days"]["2026-09-15"] == [0] * len(COLUMNS) + assert payload["daily_models"]["2026-09-14"]["model/free"][9] == 0 + assert payload["daily_models"]["2026-09-16"]["_other"][0] == 1 + assert "today/not-published" not in encoded.content.decode() + _assert_partitions(payload) + assert prior_version_from_frame(encoded.frame) == encoded.as_prior_version() + + +def test_reported_estimated_and_missing_usage_stay_separate() -> None: + payload = _payload(_encode(_aggregate([_outcome()]))) + row = payload["days"][WEEK.isoformat()] + assert row == [1, 1, 0, 0, 0, 10, 4, 0, 0, 25, 0, 1, 0, 0, 0, 0, 1, 1, 0, 0] + _assert_partitions(payload) + + +def test_measured_cohort_cannot_be_inferred_from_separate_source_counts() -> None: + measured = _outcome(output_source="reported", output_tokens=20) + missing = _outcome( + input_source="missing", + input_tokens=0, + output_source="missing", + output_tokens=0, + ) + together = _aggregate([measured, missing]).days[0].values + separate = ( + _aggregate( + [ + replace( + measured, + output_source="missing", + output_tokens=0, + ), + replace( + missing, + output_source="reported", + output_tokens=20, + ), + ] + ) + .days[0] + .values + ) + assert together[:18] == separate[:18] + assert together[18:] == (1, 30) + assert separate[18:] == (0, 0) + + +@pytest.mark.parametrize( + ("changes", "cohort"), + [ + ({}, (1, 30)), + ({"input_tokens": 0, "output_tokens": 0, "revenue_msats": 0}, (1, 0)), + ({"input_source": "estimated"}, (0, 0)), + ({"output_source": "estimated"}, (0, 0)), + ({"input_source": "missing"}, (0, 0)), + ({"output_source": "missing"}, (0, 0)), + ({"cache_read_source": "reported", "cache_read_input_tokens": 6}, (1, 36)), + ( + {"cache_creation_source": "reported", "cache_creation_input_tokens": 4}, + (1, 34), + ), + ({"cache_read_source": "estimated"}, (0, 0)), + ({"cache_creation_source": "estimated"}, (0, 0)), + ({"cache_read_input_tokens": 6}, (0, 0)), + ({"cache_creation_input_tokens": 4}, (0, 0)), + ({"completed_requests": 3, "input_tokens": 30, "output_tokens": 60}, (3, 90)), + ], +) +def test_measured_cohort_uses_matching_request_and_token_totals( + changes: dict[str, Any], cohort: tuple[int, int] +) -> None: + measured = _outcome(output_source="reported", output_tokens=20) + payload = _payload(_encode(_aggregate([replace(measured, **changes)]))) + assert payload["days"][WEEK.isoformat()][18:] == list(cohort) + _assert_partitions(payload) + + +@pytest.mark.parametrize("source", ["reported", "estimated", "missing"]) +def test_each_usage_source_lands_in_its_own_counters(source: str) -> None: + changes: dict[str, Any] = { + f"{name}_source": source + for name in ("input", "output", "cache_read", "cache_creation") + } + row = ( + _aggregate( + [ + _outcome( + cache_read_input_tokens=6, cache_creation_input_tokens=4, **changes + ) + ] + ) + .days[0] + .values + ) + if source == "reported": + assert row[1:5] == (1,) * 4 + assert row[18:] == (1, 24) + elif source == "estimated": + assert row[10:14] == (1,) * 4 + assert row[18:] == (0, 0) + else: + assert row[14:18] == (1,) * 4 + assert row[18:] == (0, 0) + + +def test_epoch_start_and_end_exclude_partial_and_unknown_days() -> None: + aggregate = _aggregate( + [_outcome(WEEK + timedelta(days=offset)) for offset in range(7)], + epoch=7, + epoch_coverage_start=WEEK + timedelta(days=1), + epoch_coverage_end=WEEK + timedelta(days=2), + today_utc=WEEK + timedelta(days=8), + ) + payload = _payload(_encode(aggregate)) + assert payload["complete"] is True + assert set(payload["days"]) == {"2026-09-15", "2026-09-16"} + _assert_partitions(payload) + assert ( + aggregate_ledger_week( + [], + epoch=0, + epoch_coverage_start=WEEK, + epoch_coverage_end=None, + week=WEEK, + today_utc=WEEK, + ) + is None + ) + + +def test_week_is_transport_coordinate_not_model_reporting_resolution() -> None: + early = _encode(_aggregate([_outcome()], epoch=1, epoch_coverage_end=WEEK)) + late = _encode( + _aggregate( + [_outcome(WEEK + timedelta(days=2))], + epoch=2, + epoch_coverage_start=WEEK + timedelta(days=2), + ) + ) + coordinate = f"38421:{early.pubkey}:provider" + digest = hashlib.sha256(coordinate.encode()).hexdigest()[:16] + assert early.d_tag == f"routstr.analytics.v2:{digest}:week:2026-09-14:epoch:1" + assert late.d_tag != early.d_tag + assert early.tags[1] == ("a", coordinate) + assert not (_payload(early)["days"].keys() & _payload(late)["days"].keys()) + + +def test_corrected_closed_week_retains_daily_model_detail() -> None: + prior = _encode(_aggregate([_outcome()], today_utc=WEEK + timedelta(days=7))) + corrected = _encode( + _aggregate( + [_outcome(model="model/corrected")], + today_utc=WEEK + timedelta(days=7), + prior_version=prior.as_prior_version(), + correction=True, + ), + created_at=prior.created_at + 1, + ) + payload = _payload(corrected) + assert payload["corrected"] is True + assert payload["corrects"] == prior.event_id + assert "model/corrected" in payload["daily_models"][WEEK.isoformat()] + assert "model/a" not in payload["daily_models"][WEEK.isoformat()] + _assert_partitions(payload) + assert ( + prior_version_from_frame(corrected.frame).daily_models == corrected.daily_models + ) + + +def test_ordinary_successor_cannot_rewrite_model_mix_or_prior_daily_totals() -> None: + prior = _encode(_aggregate([_outcome()], today_utc=WEEK + timedelta(days=1))) + for changed in (_outcome(model="changed"), _outcome(revenue_msats=100)): + aggregate = _aggregate( + [changed], + today_utc=WEEK + timedelta(days=2), + prior_version=prior.as_prior_version(), + ) + with pytest.raises(AnalyticsV2Error, match="changed a published row"): + _encode(aggregate, created_at=prior.created_at + 1) + unchanged = _aggregate( + [_outcome()], + today_utc=WEEK + timedelta(days=2), + prior_version=prior.as_prior_version(), + ) + assert ( + _payload(_encode(unchanged, created_at=prior.created_at + 1))["through"] + == "2026-09-15" + ) + + +def test_new_epochs_do_not_accept_a_different_epoch_as_predecessor() -> None: + prior = _encode(_aggregate([_outcome()], epoch=1, epoch_coverage_end=WEEK)) + fresh = _aggregate( + [], + epoch=2, + epoch_coverage_start=WEEK + timedelta(days=2), + prior_version=prior.as_prior_version(), + ) + with pytest.raises(AnalyticsV2Error, match="different coordinate"): + _encode(fresh, created_at=prior.created_at + 1) + + +def test_single_request_models_are_named_without_an_arbitrary_model_cap() -> None: + aggregate = _aggregate([_outcome(model=f"model/{index}") for index in range(30)]) + payload = _payload(_encode(aggregate)) + assert len(payload["daily_models"][WEEK.isoformat()]) == 31 + _assert_partitions(payload) + + +def test_utf8_size_folding_preserves_every_day_and_metric() -> None: + aggregate = _aggregate( + [ + _outcome( + WEEK + timedelta(days=index % 4), + "模型/" + str(index) + "/" + "x" * 120, + output_source="reported" if index % 3 else "estimated", + cache_read_source="reported", + cache_read_input_tokens=6, + cache_creation_source="reported", + cache_creation_input_tokens=4, + ) + for index in range(120) + ] + ) + full = _encode(aggregate) + assert len(full.frame) < DEFAULT_MAX_FRAME_BYTES + limit = len(full.frame) // 3 + folded = _encode(aggregate, max_frame_bytes=limit) + assert len(folded.frame) <= limit + assert any( + row["_other"][0] > 0 for row in _payload(folded)["daily_models"].values() + ) + assert any( + row["_other"][18] > 0 and row["_other"][19] > 0 + for row in _payload(folded)["daily_models"].values() + ) + assert _payload(folded)["days"] == _payload(full)["days"] + _assert_partitions(_payload(folded)) + assert prior_version_from_frame(folded.frame) == folded.as_prior_version() + with pytest.raises(FrameTooLargeError): + _encode(aggregate, max_frame_bytes=100) + + +def test_sdk_signature_and_exact_canonical_frame_round_trip() -> None: + encoded = _encode(_aggregate([_outcome(model='model/雪"\\')])) + event = encoded.event + assert Event.from_json(json.dumps(event)).verify() + assert ( + encoded.frame + == json.dumps( + ["EVENT", event], separators=(",", ":"), ensure_ascii=False + ).encode() + ) + assert prior_version_from_frame(encoded.frame) == encoded.as_prior_version() + with pytest.raises(AnalyticsV2Error, match="not canonical"): + prior_version_from_frame(json.dumps(["EVENT", event]).encode()) + for field, value in ( + ("created_at", event["created_at"] + 1), + ("content", event["content"].replace("model/", "renamed/")), + ): + tampered = {**event, field: value} + assert Event.from_json(json.dumps(tampered)).verify_signature() + with pytest.raises(AnalyticsV2Error): + prior_version_from_frame( + json.dumps( + ["EVENT", tampered], separators=(",", ":"), ensure_ascii=False + ).encode() + ) + event["sig"] = "00" * 64 + with pytest.raises(AnalyticsV2Error, match="signature"): + prior_version_from_frame( + json.dumps( + ["EVENT", event], separators=(",", ":"), ensure_ascii=False + ).encode() + ) + + +@pytest.mark.parametrize( + "change", + [ + "missing_day", + "missing_models", + "nonpartition", + "provenance", + "unsafe_integer", + "bool", + "wrong_epoch", + "draft_columns", + ], +) +def test_signed_invalid_payloads_are_rejected(change: str) -> None: + encoded = _encode(_aggregate([_outcome(output_source="reported")])) + payload = _payload(encoded) + day = WEEK.isoformat() + if change == "missing_day": + del payload["days"][day] + elif change == "missing_models": + del payload["daily_models"][day] + elif change == "nonpartition": + payload["daily_models"][day]["_other"][9] = 1 + elif change == "provenance": + payload["days"][day][10] = 1 + elif change == "unsafe_integer": + payload["days"][day][9] = MAX_SAFE_INTEGER + 1 + elif change == "bool": + payload["days"][day][9] = False + elif change == "draft_columns": + payload["columns"] = payload["columns"][:18] + payload["days"] = {day: values[:18] for day, values in payload["days"].items()} + payload["daily_models"] = { + day: {model: values[:18] for model, values in models.items()} + for day, models in payload["daily_models"].items() + } + else: + payload["epoch"] = 42 + with pytest.raises(AnalyticsV2Error): + prior_version_from_frame(_resign(encoded, payload)) + + +@pytest.mark.parametrize( + ("column", "value", "message"), + [ + (18, 2, "Measured requests"), + (19, 15, "Measured tokens"), + (18, 0, "Measured tokens"), + ], +) +def test_signed_invalid_measured_cohorts_are_rejected_even_when_models_partition( + column: int, value: int, message: str +) -> None: + encoded = _encode(_aggregate([_outcome(output_source="reported")])) + payload = _payload(encoded) + day = WEEK.isoformat() + payload["days"][day][column] = value + payload["daily_models"][day]["model/a"][column] = value + with pytest.raises(AnalyticsV2Error, match=message): + prior_version_from_frame(_resign(encoded, payload)) + + +def test_aggregate_rejects_unsafe_values_and_invalid_model_partition() -> None: + with pytest.raises(AnalyticsV2Error): + _aggregate([_outcome(revenue_msats=MAX_SAFE_INTEGER + 1)]) + with pytest.raises(AnalyticsV2Error): + _aggregate([_outcome(input_source="guessed")]) + aggregate = _aggregate([_outcome()]) + broken = replace( + aggregate, days=(DayAggregate(WEEK, (0,) * len(COLUMNS)), *aggregate.days[1:]) + ) + with pytest.raises(AnalyticsV2Error, match="partition"): + _encode(broken) + + +@pytest.mark.parametrize("provider_d", ["", "x" * 65, "node\nname"]) +def test_provider_identifier_contract(provider_d: str) -> None: + with pytest.raises(AnalyticsV2Error): + build_analytics_address( + Keys.parse(PRIVATE_KEY).public_key().to_hex(), provider_d, WEEK + ) + + +def test_shared_frontend_fixture_has_a_valid_signature_and_daily_partitions() -> None: + fixture = ( + Path(__file__).parents[1] + / "fixtures" + / "analytics-v2" + / "daily-models-signed.json" + ) + event = json.loads(fixture.read_text()) + frame = json.dumps( + ["EVENT", event], separators=(",", ":"), ensure_ascii=False + ).encode() + parsed = prior_version_from_frame(frame) + assert parsed.epoch == 3 + assert parsed.through == WEEK + timedelta(days=3) + _assert_partitions(json.loads(event["content"])) + + +def test_size_folding_an_ordinary_extension_marks_changed_model_detail_as_correction() -> ( + None +): + rows = [_outcome(model=f"model/{index}/" + "x" * 80) for index in range(30)] + first = _encode( + _aggregate(rows, today_utc=WEEK + timedelta(days=1)), + max_frame_bytes=4_000, + ) + extended_rows = rows + [ + replace(row, terminal_day=WEEK + timedelta(days=1)) for row in rows + ] + extended = _encode( + _aggregate( + extended_rows, + today_utc=WEEK + timedelta(days=2), + prior_version=first.as_prior_version(), + ), + max_frame_bytes=4_000, + created_at=first.created_at + 1, + ) + assert ( + _payload(first)["daily_models"][WEEK.isoformat()] + != _payload(extended)["daily_models"][WEEK.isoformat()] + ) + assert _payload(extended)["corrects"] == first.event_id + _assert_partitions(_payload(extended))