fix: preserve Qwen token 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.

.github/workflows/ci_python.yml

@@ -84,6 +84,14 @@ jobs:
       - name: Set up uv
         uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8
 
+      - name: Set up Node.js for reducer parity
+        if: matrix.platform == 'linux-amd64' && matrix.python-version == '3.12'
+        uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v6
+        with:
+          node-version: 24
+          cache: npm
+          cache-dependency-path: adapter-contract/typescript/package-lock.json
+
       - name: Cache cargo build
         uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1
         with:
@@ -115,6 +123,8 @@ jobs:
           npm run build --prefix adapters/typescript --workspace nemo-fabric-adapters-pi
 
       - name: Run pytest
+        env:
+          FABRIC_REQUIRE_TYPESCRIPT_REDUCER_PARITY: "1"
         run: |
           set -euo pipefail
           just --set no_uv true test-python

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

@@ -601,6 +601,136 @@ 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:
+        if self.models_exhaustive and not self.models and self.observed is not None:
+            for dimension in UsageDimension:
+                if dimension is UsageDimension.COST_USD:
+                    continue
+                count = getattr(self.observed, dimension.value)
+                if count is not None and count > 0:
+                    raise ContractValidationError(
+                        "exhaustive empty model list conflicts with positive tokens"
+                    )
+        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."""