feat: report Pi model usage per invocation

.agents/skills/contribute-adapter/SKILL.md

@@ -57,7 +57,11 @@ Keep descriptor claims, implementation, focused tests, public documentation,
 catalog entries, and packaged metadata synchronized. Start with the narrowest
 truthful capability set.
 
-For usage changes, verify invocation-local accounting through the adapter contract and consumer SDK, including warm-session deltas, unknown counters, cache semantics, and unsuccessful results. Keep generated schemas and language bindings in parity. Preserve the closed `fabric.adapter/v1alpha2` `AgentUsage` shape; carry detailed cache-write, reasoning, peak request, and per-model data in the descriptor-declared `usage.extensions["nemo_fabric.detailed_token_usage.v1"]` object until a new adapter contract revision negotiates those fields. Consumers treat any detailed field for an input or output dimension as selecting the detailed counters for that dimension, so require complete, internally consistent counters instead of relying on the top-level view as a fallback.
+For usage changes, verify invocation-local accounting through the adapter contract and consumer SDK, including warm-session deltas, omitted fields, cache semantics and unsuccessful results. Keep generated schemas and language bindings in parity. Preserve the closed `fabric.adapter/v1alpha2` `AgentUsage` shape.
+
+For a collector using shared accounting, declare `usage.extensions["nemo_fabric.usage_accounting.v1"]` in the descriptor and use Python `usage_from_observations` or TypeScript `usageFromObservations`. Keep source parsing, deduplication and invocation ownership in the collector; core normalizes counts and applies qualified pricing. An omitted count stays unknown and a reported zero stays zero. Freeze observations at return; late usage cannot move to a later invocation. An execution failure alone does not erase settled usage.
+
+The shared helper also supplies the legacy `nemo_fabric.detailed_token_usage.v1` projection for older consumers. Keep its descriptor declaration and inclusion flags consistent. Detailed legacy counters select their dimension without falling back to top-level counters. Do not replace incomplete shared accounting with legacy fields, logs or Relay totals, and do not promote client estimates to provider-reported cost.
 
 For `instructions.system`, keep `config.system_instruction_modes`, planning
 behavior, direct adapter validation, and target-native composition synchronized.

adapter-contract/python/src/nemo_fabric_adapter_contract/models.py

@@ -521,6 +521,127 @@ def _validate(self) -> None:
             )
 
 
+class UsageDimension(StrEnum):
+    """Raw dimensions in the declared usage-accounting extension."""
+
+    INPUT_TOKENS = "input_tokens"
+    CACHED_INPUT_TOKENS = "cached_input_tokens"
+    CACHE_WRITE_INPUT_TOKENS = "cache_write_input_tokens"
+    OUTPUT_TOKENS = "output_tokens"
+    REASONING_TOKENS = "reasoning_tokens"
+    TOTAL_TOKENS = "total_tokens"
+    PEAK_REQUEST_INPUT_TOKENS = "peak_request_input_tokens"
+    COST_USD = "cost_usd"
+
+
+class GapReason(StrEnum):
+    """Why complete accounting is unavailable."""
+
+    UNFINISHED_WORK = "unfinished_work"
+    MISSING_USAGE = "missing_usage"
+    SCOPE_UNVERIFIED = "scope_unverified"
+
+
+@dataclass(slots=True, kw_only=True)
+class ReportedUsage(ContractModel):
+    """Raw provider/harness counts, retaining independent field presence."""
+
+    input_tokens: int | None = _optional()
+    cached_input_tokens: int | None = _optional()
+    cache_write_input_tokens: int | None = _optional()
+    input_tokens_include_cache: bool | None = _optional()
+    output_tokens: int | None = _optional()
+    reasoning_tokens: int | None = _optional()
+    output_tokens_include_reasoning: bool | None = _optional()
+    total_tokens: int | None = _optional()
+    peak_request_input_tokens: int | None = _optional()
+    cost_usd: float | None = _optional()
+
+    def _validate(self) -> None:
+        import math
+
+        for dimension in UsageDimension:
+            value = getattr(self, dimension.value)
+            if dimension is not UsageDimension.COST_USD:
+                _bounded_int(value, dimension.value, (1 << 64) - 1)
+            elif value is not None and (
+                not math.isfinite(value) or value < 0 or value * 1e9 >= 1 << 64
+            ):
+                raise ContractValidationError(
+                    "cost must fit finite nonnegative nano-USD"
+                )
+
+
+@dataclass(slots=True, kw_only=True)
+class UsageGap(ContractModel):
+    """A nonempty set of incomplete raw dimensions."""
+
+    dimensions: list[UsageDimension]
+    reason: GapReason
+
+    def _validate(self) -> None:
+        if not self.dimensions or len(set(self.dimensions)) != len(self.dimensions):
+            raise ContractValidationError("gap dimensions must be nonempty and unique")
+
+
+@dataclass(slots=True, kw_only=True)
+class UsageRequest(ContractModel):
+    """One deduplicated model-call snapshot, not an aggregate."""
+
+    usage: ReportedUsage
+    gap: GapReason | None = _optional()
+    tokens_aligned: bool = False
+
+
+@dataclass(slots=True, kw_only=True)
+class AccountedModel(ContractModel):
+    """Complete and observed evidence for one literal model/provider identity."""
+
+    model: str
+    gaps: list[UsageGap]
+    provider: str | None = _optional()
+    reported: ReportedUsage = field(default_factory=ReportedUsage)
+    observed: ReportedUsage | None = _optional()
+
+    observed_tokens_aligned: bool = False
+    requests: list[UsageRequest] = _empty_list()
+    requests_exhaustive: bool = False
+
+    def _validate(self) -> None:
+        _nonblank(self.model, "model")
+        if self.requests_exhaustive and not self.requests:
+            raise ContractValidationError("exhaustive requests must be nonempty")
+        if self.provider is not None:
+            _nonblank(self.provider, "provider")
+        for gap in self.gaps:
+            if any(
+                getattr(self.reported, dimension.value) is not None
+                for dimension in gap.dimensions
+            ):
+                raise ContractValidationError(
+                    "complete value conflicts with explicit gap"
+                )
+
+
+@dataclass(slots=True, kw_only=True)
+class UsageAccounting(ContractModel):
+    """Typed nemo_fabric.usage_accounting.v1 payload.
+
+    Declare its schema in extension_schemas.usage. This does not change the
+    closed AgentUsage contract. Per-model complete counts live in reported.
+    """
+
+    models_exhaustive: bool
+    gaps: list[UsageGap]
+    observed: ReportedUsage | None = _optional()
+    models: list[AccountedModel] = _empty_list()
+
+    def _validate(self) -> None:
+        identities = [(model.provider, model.model) for model in self.models]
+        if len(set(identities)) != len(identities):
+            raise ContractValidationError("duplicate model identity")
+
+
 @dataclass(slots=True, kw_only=True)
 class AgentRunResult(AgentContractBlock):
     """Southbound terminal result returned by an adapter target."""

adapter-contract/python/src/nemo_fabric_adapter_contract/usage.py

@@ -0,0 +1,238 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared raw-evidence reduction for Python collectors.
+
+Normalization, aggregate/model reconciliation and pricing belong to core.
+Collectors supply settled snapshots and explicit coverage gaps, never inferred
+zero usage from HTTP status. One record represents one deduplicated source unit.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from collections.abc import Iterable
+from dataclasses import dataclass
+import math
+
+from nemo_fabric_adapter_contract.models import AccountedModel
+from nemo_fabric_adapter_contract.models import AgentUsage
+from nemo_fabric_adapter_contract.models import GapReason
+from nemo_fabric_adapter_contract.models import ReportedUsage
+from nemo_fabric_adapter_contract.models import UsageAccounting
+from nemo_fabric_adapter_contract.models import UsageDimension
+from nemo_fabric_adapter_contract.models import UsageGap
+from nemo_fabric_adapter_contract.models import UsageRequest
+
+ACCOUNTING_KEY = "nemo_fabric.usage_accounting.v1"
+DETAIL_KEY = "nemo_fabric.detailed_token_usage.v1"
+
+
+@dataclass(frozen=True, kw_only=True)
+class UsageObservation:
+    """One source unit's deduplicated raw snapshot and completeness boundary."""
+
+    reported: ReportedUsage
+    model: str | None = None
+    provider: str | None = None
+    gap: GapReason | None = None
+    tokens_aligned: bool = False
+    is_model_call: bool = False
+    gaps: tuple[UsageGap, ...] = ()
+
+
+def _reduce(
+    records: list[UsageObservation],
+) -> tuple[ReportedUsage, ReportedUsage, list[UsageGap]]:
+    complete: dict[str, object] = {}
+    observed: dict[str, object] = {}
+    gaps: dict[GapReason, list[UsageDimension]] = defaultdict(list)
+    for dimension in UsageDimension:
+        key = dimension.value
+        values = [getattr(record.reported, key) for record in records]
+        known = [value for value in values if value is not None]
+        total = None
+        if known:
+            if dimension is UsageDimension.COST_USD:
+                amount = math.fsum(known)
+                if math.isfinite(amount) and 0 <= amount * 1e9 < 1 << 64:
+                    total = amount
+            elif dimension is UsageDimension.PEAK_REQUEST_INPUT_TOKENS:
+                total = max(known)
+            else:
+                amount = sum(known)
+                if amount <= (1 << 64) - 1:
+                    total = amount
+        if total is not None:
+            observed[key] = total
+        causes = {record.gap for record in records if record.gap is not None}
+        causes.update(
+            gap.reason
+            for record in records
+            for gap in record.gaps
+            if dimension in gap.dimensions
+        )
+        if len(known) != len(records):
+            causes.add(GapReason.MISSING_USAGE)
+        if known and total is None:
+            causes.add(GapReason.SCOPE_UNVERIFIED)
+        if total is not None and not causes:
+            complete[key] = total
+        elif not causes:
+            causes.add(GapReason.SCOPE_UNVERIFIED if not records else GapReason.MISSING_USAGE)
+        for cause in sorted(causes):
+            gaps[cause].append(dimension)
+    for flag, counter in (
+        ("input_tokens_include_cache", "input_tokens"),
+        ("output_tokens_include_reasoning", "output_tokens"),
+    ):
+        flags = {
+            getattr(record.reported, flag)
+            for record in records
+            if getattr(record.reported, counter) is not None
+        }
+        if len(flags) == 1 and None not in flags:
+            observed[flag] = next(iter(flags))
+            if counter in complete:
+                complete[flag] = observed[flag]
+        elif len(
+            [
+                record
+                for record in records
+                if getattr(record.reported, counter) is not None
+            ]
+        ) == 1 and flags == {None}:
+            # One raw source remains useful even when its convention is unknown.
+            # Core withholds normalized values; do not discard literal evidence.
+            pass
+        elif counter in observed:
+            # Mixed conventions cannot be combined as one raw input/output count.
+            complete.pop(counter, None)
+            observed.pop(counter, None)
+            dimension = UsageDimension(counter)
+            if not all(
+                record.is_model_call and record.model is not None for record in records
+            ):
+                if dimension not in gaps[GapReason.SCOPE_UNVERIFIED]:
+                    gaps[GapReason.SCOPE_UNVERIFIED].append(dimension)
+    return (
+        ReportedUsage.from_mapping(complete),
+        ReportedUsage.from_mapping(observed),
+        [
+            UsageGap(dimensions=dimensions, reason=reason)
+            for reason, dimensions in sorted(gaps.items())
+        ],
+    )
+
+
+def _observed_tokens_aligned(records: list[UsageObservation]) -> bool:
+    # Shape alone never proves scope. The collector must first identify a
+    # coherent source snapshot; matching presence then keeps reductions aligned.
+    fields = [
+        "input_tokens",
+        "cached_input_tokens",
+        "cache_write_input_tokens",
+        "output_tokens",
+        "peak_request_input_tokens",
+    ]
+    if any(r.reported.output_tokens_include_reasoning is False for r in records):
+        fields.append("reasoning_tokens")
+    relevant = [
+        r
+        for r in records
+        if any(getattr(r.reported, field) is not None for field in fields)
+    ]
+    return (
+        bool(relevant)
+        and all(r.tokens_aligned and not r.gaps for r in relevant)
+        and all(
+            len({getattr(r.reported, field) is not None for r in relevant}) == 1
+            for field in fields
+        )
+    )
+
+
+def _request_usage(record: UsageObservation) -> ReportedUsage:
+    # A scoped observation is not a complete per-call count for that dimension.
+    # Keep the literal value in model observations without allowing requests to
+    # re-certify it or use it as a complete pricing component.
+    fields = record.reported.to_mapping()
+    for gap in record.gaps:
+        for dimension in gap.dimensions:
+            fields.pop(dimension.value, None)
+    return ReportedUsage.from_mapping(fields)
+
+
+def usage_from_observations(records: Iterable[UsageObservation]) -> AgentUsage:
+    """Reduce snapshots once, preserving field presence and independently known values.
+
+    Empty source coverage is unverified rather than assumed to mean zero calls.
+    A collector that proves no model work occurred can report an explicit zero
+    snapshot. This helper does not change scheduling or wait for late work.
+    """
+
+    records = list(records)
+    reported, observed, gaps = _reduce(records)
+    groups: dict[tuple[str | None, str], list[UsageObservation]] = defaultdict(list)
+    for record in records:
+        if record.model is not None:
+            groups[(record.provider, record.model)].append(record)
+    models = []
+    for (provider, model), group in sorted(
+        groups.items(), key=lambda item: (item[0][0] or "", item[0][1])
+    ):
+        complete_model, observed_model, model_gaps = _reduce(group)
+        models.append(
+            AccountedModel(
+                model=model,
+                provider=provider,
+                reported=complete_model,
+                observed=observed_model,
+                observed_tokens_aligned=_observed_tokens_aligned(group),
+                gaps=model_gaps,
+                requests=[
+                    UsageRequest(
+                        usage=_request_usage(record),
+                        gap=record.gap,
+                        tokens_aligned=record.tokens_aligned,
+                    )
+                    for record in group
+                ]
+                if all(record.is_model_call for record in group)
+                else [],
+                requests_exhaustive=all(record.is_model_call for record in group),
+            )
+        )
+    accounting = UsageAccounting(
+        models_exhaustive=bool(records)
+        and all(record.model is not None and record.gap is None for record in records),
+        gaps=gaps,
+        observed=observed,
+        models=models,
+    )
+    base_keys = {
+        "input_tokens",
+        "cached_input_tokens",
+        "input_tokens_include_cache",
+        "output_tokens",
+        "total_tokens",
+        "cost_usd",
+    }
+    complete_mapping = reported.to_mapping()
+    detail_keys = {
+        "input_tokens",
+        "cached_input_tokens",
+        "cache_write_input_tokens",
+        "input_tokens_include_cache",
+        "output_tokens",
+        "reasoning_tokens",
+        "output_tokens_include_reasoning",
+        "peak_request_input_tokens",
+    }
+    detail = {
+        key: value for key, value in complete_mapping.items() if key in detail_keys
+    }
+    return AgentUsage(
+        **{key: value for key, value in complete_mapping.items() if key in base_keys},
+        extensions={ACCOUNTING_KEY: accounting.to_mapping(), DETAIL_KEY: detail},
+    )

adapter-contract/typescript/package.json

@@ -26,9 +26,10 @@
     "build": "node scripts/clean.mjs && tsc -p tsconfig.build.json",
     "test:generator": "node --test test/*.test.mjs",
     "test:types": "tsc -p test/tsconfig.json --noEmit",
+    "test:usage": "tsc -p tsconfig.build.json && node --test test/usage-reducer.test.js",
     "test:dependencies": "node scripts/check-dependencies.mjs && npm ls --all && npm audit --audit-level=high",
     "pack:check": "npm run build && node scripts/check-package.mjs",
-    "test": "npm run generate:check && npm run test:generator && npm run test:types && npm run test:dependencies && npm run pack:check",
+    "test": "npm run generate:check && npm run test:generator && npm run test:types && npm run test:usage && npm run test:dependencies && npm run pack:check",
     "prepack": "npm run generate:check && npm run build"
   },
   "publishConfig": {