diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 49aa0c531c..2cb61b940e 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -453,7 +453,7 @@ ClawMetry is open-core. `clawmetry/entitlements.py` is the single source of truth for what an install may do, and for the runtime and channel catalogues that every count in this repo is derived from. `clawmetry/license.py` verifies self-hosted license keys offline with Ed25519, using the `cryptography` -dependency rather than a new one. `routes/entitlement.py` exposes the resolved +dependency rather than a new one. `routes/entitlement/` (a package: init re-exports flat namespace, shared helpers module, eight endpoint files with 434 handlers) exposes the resolved entitlement at `/api/entitlement*`. The resolver runs in **GRACE** mode until the announced enforce date: every diff --git a/CLAUDE.md b/CLAUDE.md index 0659734329..b0a53c2e7a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -21,7 +21,7 @@ See `ARCHITECTURE.md` for the full deep dive, and `docs/MODULE_MAP.md` (generate `docs/MODULE_MAP.md` is the **generated** inventory: every module, the blueprints it defines, the URL space it owns, and a coarse size band. `scripts/gen_module_map.py` regenerates it and CI fails when it drifts. The tables below are a short curated index of what you reach for most often, deliberately without line counts (they went stale within weeks every time they were written down). -**Five files are big enough to change how you work on them**: `routes/entitlement.py` (~48k lines), `clawmetry/entitlements.py` (~31k), `clawmetry/sync.py` (~26k), `dashboard.py` (~21k), `clawmetry/local_store.py` (~20k). Drift Bot reads only the head of a long file, so anything added deep inside one is reported as "not implemented" forever. Put new capability in a new short module and re-export it, rather than appending 300 lines to a 20k-line file. +**Five files are big enough to change how you work on them**: `routes/entitlement/` (~48k lines, now a package: a re-export init, a shared helpers module, and eight endpoint files), `clawmetry/entitlements.py` (~31k), `clawmetry/sync.py` (~26k), `dashboard.py` (~21k), `clawmetry/local_store.py` (~20k). Drift Bot reads only the head of a long file, so anything added deep inside one is reported as "not implemented" forever. Put new capability in a new short module and re-export it, rather than appending 300 lines to a 20k-line file. ### Core | File | Purpose | @@ -50,7 +50,7 @@ All HTTP endpoints live here, organised by feature: 70 modules, 82 blueprints, l | `routes/hooks.py` | `bp_hooks` — hook install / status / uninstall per runtime, and the gate's decision log | | `routes/infra.py` | `bp_logs` + `bp_memory` + `bp_security` + `bp_config` — logs stream, memory files, security posture, cost-optimizer | | `routes/meta.py` | `bp_auth` + `bp_gateway` + `bp_otel` + `bp_version` + `bp_version_impact` + `bp_cloud_relay` + `bp_otlp_traces` — auth, gateway proxy, OTLP ingestion, version meta | -| `routes/entitlement.py` | `bp_entitlement` — the resolved entitlement plus the preview / diff / batch family at `/api/entitlement*` | +| `routes/entitlement/` | `bp_entitlement` — the resolved entitlement plus the preview / diff / batch family at `/api/entitlement*` (package: init re-exports flat namespace, shared helpers module, eight endpoint files with 434 handlers) | | `routes/alerts.py` | `bp_alerts` + `bp_budget` — alert rules, webhooks, velocity, budget config | | `routes/crons.py` | `bp_crons` — cron CRUD + run log + health summary | | `routes/signals.py` | `bp_signals` — Behaviour Signals read API: `/api/signals` (rate, count, eligible turns, trend, by model and runtime, coverage, plain-words headline), `/api/signals//sessions` (sessions, never phrases) | diff --git a/docs/MODULE_MAP.md b/docs/MODULE_MAP.md index 6b66d2fb97..dd303b630f 100644 --- a/docs/MODULE_MAP.md +++ b/docs/MODULE_MAP.md @@ -4,7 +4,7 @@ > `python3 scripts/gen_module_map.py` (CI fails on drift via > `tests/test_module_map_drift.py`). -239 modules, 81 Flask blueprints. `CLAUDE.md` carries a short curated table of the ones you reach for most often; this is the whole list. +247 modules, 81 Flask blueprints. `CLAUDE.md` carries a short curated table of the ones you reach for most often; this is the whole list. Size bands are deliberately coarse so this file does not churn on every PR: **small** is under 200 lines, **medium** under 1k, **large** under 5k, **huge** is 5k and up. @@ -46,7 +46,6 @@ One module per feature, each owning one or more Flask blueprints. New endpoints | `routes/delegated.py` | small | `bp_delegated` | `/api/cursor`, `/api/delegated-usage` | Connect a Cursor account from the dashboard, and read delegated usage. | | `routes/device.py` | medium | `bp_device` | `/api/device`, `/device-preview` | Device snapshot — a compact, screen-sized JSON for hardware companions. | | `routes/dives.py` | medium | `bp_dives` | `/api/dives` | ClawMetry Dives: NL-to-SQL-to-chart over the local DuckDB store. | -| `routes/entitlement.py` | huge | `bp_entitlement` | `/api/entitlement`, `/api/features`, `/api/license`, `/api/paywall`, `/api/runtimes`, `/api/tiers` | ``bp_entitlement``. | | `routes/evals.py` | medium | `bp_evals` | `/api/evals`, `/api/evaluators` | Eval (LLM-as-judge) endpoints. | | `routes/extensions.py` | small | `bp_extensions` | `/api/extensions` | diagnostic introspection for the entry-point plugin loader. | | `routes/fleet_history.py` | medium | `bp_fleet` | `/api/nodes`, `/fleet` | Multi-node fleet endpoints. | @@ -97,6 +96,22 @@ One module per feature, each owning one or more Flask blueprints. New endpoints | `routes/usage.py` | huge | `bp_usage` | `/api/activity-today`, `/api/anomalies`, `/api/efficiency`, `/api/forward-progress`, `/api/model-attribution`, `/api/nemo-cap-status`, `/api/runtime-summary`, `/api/sessions`, `/api/skill-attribution`, `/api/skills`, `/api/token-attribution`, `/api/token-velocity`, `/api/usage` | Usage / analytics / anomaly / attribution endpoints. | | `routes/workspaces.py` | small | `bp_workspaces` | `/api/workspaces` | Multi-profile OpenClaw workspace discovery + switcher. | +## Entitlement route package (`routes/entitlement/`) + +The entitlement API surface, split from a single 47k-line file into a package: a shared helpers module holds the blueprint and helpers; eight endpoint files hold the 434 route handlers. + +| Module | Size | Blueprints | Serves | Purpose | +|---|---|---|---|---| +| `routes/entitlement/_endpoints_01.py` | large | | `/api/entitlement` | endpoint handlers api_entitlement .. | +| `routes/entitlement/_endpoints_02.py` | large | | `/api/entitlement` | endpoint handlers api_entitlement_missing_features_at_path .. | +| `routes/entitlement/_endpoints_03.py` | large | | `/api/entitlement`, `/api/features`, `/api/license`, `/api/runtimes`, `/api/tiers` | endpoint handlers api_entitlement_min_tier .. | +| `routes/entitlement/_endpoints_04.py` | large | | `/api/entitlement`, `/api/license`, `/api/paywall` | endpoint handlers api_license_state_at .. | +| `routes/entitlement/_endpoints_05.py` | large | | `/api/entitlement` | endpoint handlers api_entitlement_next_tier_spec .. | +| `routes/entitlement/_endpoints_06.py` | large | | `/api/entitlement` | endpoint handlers api_entitlement_feature_catalog_at_path_batch .. | +| `routes/entitlement/_endpoints_07.py` | large | | `/api/entitlement` | endpoint handlers api_entitlement_lock_reason_at_path .. | +| `routes/entitlement/_endpoints_08.py` | large | | `/api/entitlement` | endpoint handlers api_entitlement_min_tier_for_features_batch .. | +| `routes/entitlement/_shared.py` | huge | `bp_entitlement` | | imports, constants, the blueprint and every non-handler helper the endpoint modules call. | + ## Shared helpers (`helpers/`) Helpers extracted out of `dashboard.py`. Route modules still reach the ones that have not moved yet via a late `import dashboard as _d`. diff --git a/docs/ci_test_coverage_baseline.json b/docs/ci_test_coverage_baseline.json index 0a971624bf..4342b80e0a 100644 --- a/docs/ci_test_coverage_baseline.json +++ b/docs/ci_test_coverage_baseline.json @@ -8,6 +8,6 @@ "Related: issue #5813" ], "total": 1143, - "listed": 212, - "unlisted_max": 931 + "listed": 213, + "unlisted_max": 930 } diff --git a/routes/entitlement.py b/routes/entitlement.py deleted file mode 100644 index d306dcbbaf..0000000000 --- a/routes/entitlement.py +++ /dev/null @@ -1,47708 +0,0 @@ -""" -routes/entitlement.py -- ``bp_entitlement``. - -Exposes the resolved open-core entitlement so the frontend knows which -runtimes/features to surface (and, once enforcement is live, which to render -locked behind an upgrade CTA). Backed by :mod:`clawmetry.entitlements`, which -is the single source of truth -- handlers never re-derive tier logic here. - - GET /api/entitlement -- the current Entitlement as JSON. - GET /api/entitlement/diagnostic -- the *inputs* the resolver consulted - (license/cloud-plan presence, enforce - env, cache liveness) for operator - triage. - POST /api/entitlement/refresh -- drop the cache and return the freshly - re-resolved Entitlement. - GET /api/entitlement/required-tier -- resolve the minimum purchasable tier - for a feature=, runtime=, channels=, - or retention_days= key. The capacity - axes (channels / retention_days) wrap - the matching ``min_tier_for_*`` Python - helpers so the same endpoint answers - all four "what tier do I need" axes - off one URL. - GET /api/entitlement/lock-reason -- human-readable explanation of why a - feature=, runtime=, channels= or - retention_days= key is locked, - carrying the structured - ``required_tier`` payload alongside - the message so a paywall tooltip can - render "Locked: . [Upgrade to - ]" in one round-trip. The four - axes match the ones on - ``/api/entitlement/required-tier``. - GET /api/entitlement/upgrade-diff -- features + runtimes a target tier - would add on top of the current ent. - GET /api/entitlement/downgrade-diff -- features + runtimes a target tier - would REMOVE from the current ent. - GET /api/entitlement/tier-diff -- arbitrary-endpoint diff between any - two tiers (``?from=&to=``); - generalises ``/upgrade-diff`` / - ``/downgrade-diff`` from "current vs - target" to "any tier vs any tier" so - a "Compare A vs B" pricing-page - widget can render any pair without - first switching the resolver. - GET /api/entitlement/preview -- the full Entitlement.to_dict() shape - rendered for an arbitrary tier so the - upgrade-CTA card can show concrete - numbers without per-tier derivation - in JS. - GET /api/entitlement/required-tier-batch -- plural sibling of - ``/required-tier``: takes - ``features=a,b,c`` and/or - ``runtimes=x,y,z`` (comma-separated) - and returns the cheapest tier - admitting *all* of them at once. - Lets a dashboard answer "I'm using - fleet + otel_export + sso -- what - tier covers everything?" in a single - round-trip. - GET /api/entitlement/required-tier-breakdown -- per-axis breakdown sibling - of ``/required-tier-batch``: same - inputs and top-level - ``required_tier`` field, but - additionally exposes each axis' - individual ``min_tier`` and calls - out which axis (or axes, on a tie) - is the *binding* constraint driving - the aggregate floor -- so a paywall - CTA can render "You need Pro - *because* you have 8 channels - (Starter caps at 5)" off ONE - round-trip. - GET /api/entitlement/lock-reason-batch -- per-item plural sibling of - ``/lock-reason``: same CSV + - capacity inputs as - ``/required-tier-batch``, but - preserves per-item ``reason`` / - ``locked`` / ``required_tier`` rows - so a Settings or paywall matrix UI - renders N rows off one round-trip - instead of N calls. - GET /api/entitlement/tier-unlocks-batch -- plural sibling of - ``/tier-unlocks``: returns the full - pricing-page marginal-unlock ladder - in one pass. - GET /api/entitlement/tier-unlocks-path -- arbitrary-endpoint stepwise - unlock path between any two tiers - (``?from=&to=``); unlocks-focused - analogue of ``/tier-path`` (full - ``tier_diff`` per rung) and - ``/capacity-diff-path`` (capacity- - only per rung). Each row is a - ``tier_unlocks`` payload between - the previous step in the path and - the current rung. - GET /api/entitlement/capacity-diff-batch -- plural sibling of - ``/capacity-diff``: per-axis - capacity transitions (channels / - retention / nodes) for every - purchasable tier in one pass so - a pricing-page table can render - the capacity column off one - round-trip. - GET /api/entitlement/capacity-diff-path -- path analogue of - ``/capacity-diff-batch``: per-rung - capacity transition along an - arbitrary ``?from=&to=`` segment, - capacity-only mirror of - ``/tier-path`` so a capacity-only - pricing widget can render - channel / retention / node - marginal steps between two tiers - off one round-trip. - GET /api/entitlement/capacity-headroom-path -- path analogue of - ``/capacity-headroom-batch``: - per-rung capacity-headroom envelope - along an arbitrary ``?from=&to=`` - segment given caller-supplied - per-axis usage. Headroom-shaped - mirror of ``/capacity-diff-path`` - / ``/tier-unlocks-path`` / - ``/tier-locks-path`` / - ``/preview-path`` -- rungs line - up rung-for-rung with those four - siblings so an upgrade-walkthrough - UI can render "watch your headroom - recover rung by rung" off ONE - round-trip. - GET /api/entitlement/capacity-headroom-path-batch -- batch sibling of - ``/capacity-headroom-path`` and - headroom-shaped twin of - ``/capacity-diff-path-batch``: walk - the per-rung headroom envelopes from - ONE ``?from=`` to N candidate - ``?to=a,b,c`` destinations in ONE - round-trip. Fan-out shape matches - ``/capacity-diff-path-batch`` and - ``/tier-spec-path-batch``; unknown - destinations bucket into - ``unknown[]`` instead of 404ing. - GET /api/entitlement/next-tier-capacity-headroom -- scalar "one rung up" - sibling of ``/capacity-headroom``: - per-axis headroom envelope for the - tier immediately above the resolved - entitlement given caller-supplied - usage. Envelope shape mirrors - ``/next-tier-unlocks`` (current-tier - context + null-at-ceiling), inner - ``headroom`` matches - ``/capacity-headroom-at`` byte-for- - byte. Fills the "next-tier" slot on - the capacity-headroom axis - alongside the caps-only - ``/next-tier-capacity-diff`` and - the marginal-features - ``/next-tier-unlocks``. - GET /api/entitlement/previous-tier-capacity-headroom -- downgrade twin of - ``/next-tier-capacity-headroom``: - per-axis headroom envelope for the - tier immediately below the resolved - entitlement given caller-supplied - usage. Axes whose inner - ``over_limit`` flips ``True`` are - exactly the ones the caller would - lose headroom on. Envelope shape - matches - ``/next-tier-capacity-headroom`` - byte-for-key with ``direction`` - echoing ``"downgrade"``. - GET /api/entitlement/preview-batch -- plural sibling of ``/preview``: - the full ``Entitlement.to_dict`` - shape rendered for every purchasable - tier in one pass so a pricing-page - table can render the cumulative-state - column off one round-trip. - GET /api/entitlement/preview-path -- arbitrary-endpoint stepwise - cumulative-state path between any two - tiers (``?from=&to=``); path analogue - of ``/preview-batch`` and the - cumulative-state sibling of - ``/tier-path`` / ``/tier-unlocks-path`` - / ``/tier-locks-path`` / - ``/capacity-diff-path``. Each row is - the full ``/preview`` payload for that - rung so an upgrade-walkthrough surface - can render the "Cloud Pro: 90-day - retention, ..." card at every step - off one round-trip. - GET /api/entitlement/tier-locks -- marginal-loss companion of - ``/tier-unlocks``: features + runtimes - that disappear when you step down to - the named tier from the next-higher - purchasable tier. - GET /api/entitlement/upgrade-path -- ordered marginal-unlock ladder from - the resolved tier upward (current- - user-relative sibling of - ``/tier-unlocks-batch``). - GET /api/entitlement/downgrade-path -- ordered cumulative-loss ladder from - the resolved tier downward (direction- - flipped sibling of ``/upgrade-path``). - GET /api/entitlement/tier-path -- arbitrary-endpoint stepwise path - between any two tiers (``?from=&to=``); - path analogue of ``/tier-diff``, - generalising ``/upgrade-path`` / - ``/downgrade-path`` from "current vs - target" to "any vs any" with each - row a marginal-step ``tier_diff`` - payload. - GET /api/entitlement/affordable-tiers -- plural sibling of - ``/required-tier-batch``: returns - the full ordered list of purchasable - tiers admitting a constraint bundle - (not just the floor) so a pricing - page can render "you need at least - Starter -- Pro and Enterprise also - qualify" off one round-trip. - GET /api/entitlement/tiers-for -- inverse of ``/required-tier``: the - full ladder of tiers that grant a - ``feature=`` or ``runtime=`` key - (the "Available in: Pro, - Self-hosted Pro, Trial, Enterprise" - availability list a pricing-page - row or feature tooltip needs). - GET /api/entitlement/tiers-for-at -- hypothetical-perspective sibling of - ``/tiers-for``: same ladder scoped - by a caller-supplied - ``tier=`` so an ``_at`` - walkthrough URL is uniform across - every ``_at`` sibling. - GET /api/entitlement/tiers-for-batch-at -- hypothetical-perspective sibling - of ``/tiers-for-batch``: every - known feature + runtime in one pass - scoped by ``tier=``. - GET /api/runtimes -- the full runtime catalog. - GET /api/tiers -- the full tier ladder with per-tier metadata. - GET /api/entitlement/feature-catalog -- bare sibling of - ``/feature-catalog-at``: the resolved - feature catalogue wrapped in the same - ``{tier, features, grace, enforced}`` - envelope the ``-at`` sibling uses, so - a client hydrating every catalog - variant (bare, ``-at``, - ``-at-batch``, ``-path``, ...) can do - it off one prefix instead of mixing - ``/api/features`` with - ``/api/entitlement/feature-catalog-at``. - GET /api/entitlement/runtime-catalog -- bare sibling of - ``/runtime-catalog-at`` for the - runtime axis; same envelope shape. - GET /api/entitlement/tier-catalog -- bare sibling of - ``/tier-catalog-at`` for the tier - ladder; same envelope shape (with - the resolved tier mirrored into the - ``tier`` key to match the ``-at`` - sibling). - GET /api/entitlement/tier-spec -- scalar sibling of ``/api/tiers``: - full per-tier descriptor for one - ``tier=`` key (label, rank, - retention, channel/node limits, - features + paid runtimes carried) - so a pricing-page column / upsell - tooltip can hydrate off one - round-trip instead of walking the - full ladder client-side. - GET /api/entitlement/tier-catalog-at -- what-if sibling of the tier - ladder: returns the full - ``tier_catalog`` rows but with - ``is_current`` recomputed as if - the install were on the named - ``tier=`` instead of the live - resolved entitlement. Mirrors - ``/feature-catalog-at`` and - ``/runtime-catalog-at`` for the - tier ladder so a pricing- - comparison UI can render any - hypothetical "current tier" - without first switching the live - resolver. - GET /api/entitlement/tier-catalog-at-batch -- batch what-if sibling - of ``/tier-catalog-at``: full tier - ladders for N hypothetical source - tiers (``?tiers=a,b,c``) off ONE - call, each with ``is_current`` - flipped to its own source. Mirrors - ``/feature-catalog-at-batch`` and - ``/runtime-catalog-at-batch`` on the - tier axis so a pricing-comparison - matrix UI can render the ladder - side-by-side from every hypothetical - perspective off ONE round-trip - instead of N calls. - GET /api/entitlement/tier-spec-at -- scalar what-if sibling of - ``/tier-catalog-at``: the single - tier descriptor for ``target=`` with - ``is_current`` computed as if the - install were on ``tier=``. Lets a - pricing-comparison tooltip hydrate - against ONE tier descriptor from a - hypothetical perspective in one - round-trip instead of fetching the - full ``/tier-catalog-at`` payload. - GET /api/entitlement/tier-spec-path -- arbitrary-endpoint stepwise spec- - shaped path between any two tiers - (``?from=&to=``); path-shaped - sibling of ``/tier-spec-at-batch`` - and spec-shaped sibling of - ``/tier-path`` / ``/capacity-diff- - path`` / ``/tier-unlocks-path`` / - ``/tier-locks-path`` / ``/preview- - path``. Each row is a - ``tier_spec_at`` row pinned on - ``from=`` for ``target=``, so - the marketing-shaped descriptor - (``label``, ``is_paid``, - ``unlocks_paid_runtimes``, - ``retention_days``, - ``channel_limit``, ``node_limit``, - ``features``, ``runtimes``) hydrates - at every rung between two tiers - off one round-trip. - GET /api/entitlement/feature-catalog-path -- arbitrary-endpoint - stepwise feature-catalog path between - any two tiers (``?from=&to=``); the - full-catalog sibling of - ``/feature-spec-path`` and the path- - shaped sibling of - ``/feature-catalog-at-batch``. Each - row is a ``/feature-catalog-at`` - payload at ``rung=`` so an - upgrade-walkthrough surface hydrates - every rung's full catalogue off one - round-trip. - GET /api/entitlement/runtime-catalog-path -- runtime-axis twin of - ``/feature-catalog-path``. Together - the pair lets an upgrade-walkthrough - UI render every feature + runtime - column at every rung off two calls - instead of first walking - ``/tier-path`` and then hydrating - each rung individually. - GET /api/entitlement/tier-catalog-path -- tier-axis twin of - ``/feature-catalog-path`` / - ``/runtime-catalog-path``. Each row - is a ``/tier-catalog-at`` payload at - ``rung=`` so an upgrade- - walkthrough surface hydrates the - full pricing ladder at every rung - between two tiers off one round- - trip. Together the three - ``_catalog_path`` endpoints render - every tier + feature + runtime - column at every rung off three calls - instead of walking ``/tier-path`` - and hydrating each rung - individually. - GET /api/entitlement/runtime-detection -- pair the - :mod:`clawmetry.runtime_probe` - presence probes with the resolved - entitlement so the dashboard can - render "runtimes on this machine + - which unlock at which tier" in one - round-trip. Each probe row carries - ``found`` (present on disk), - ``allowed`` (granted by the current - tier), and the paid ``required_tier`` - to unlock it if it is not; the - envelope also carries - ``actionable_tier`` -- the single - cheapest tier that unlocks every - detected-but-locked runtime -- so - a paywall CTA does not need N - extra ``/required-tier`` calls. -""" - -from __future__ import annotations - -import logging -import os -import time - -from flask import Blueprint, jsonify, request -from routes.paywall_lifecycle import ( - PAYWALL_LIFECYCLE_EVENTS as _PAYWALL_LIFECYCLE_EVENTS, # noqa: F401 - ping_paywall_lifecycle as _ping_paywall_lifecycle, -) - -logger = logging.getLogger("clawmetry.routes.entitlement") - -bp_entitlement = Blueprint("entitlement", __name__) - - -# Last-resort minimal OSS-free snapshot used only if BOTH ``get_entitlement`` -# and ``_oss_free().to_dict()`` raise (e.g. the entitlements module itself -# fails to import). Kept in-line so this branch does not re-import the -# possibly-broken module. Keys mirror the top-level shape of -# :meth:`clawmetry.entitlements.Entitlement.to_dict` so a paywall UI reading -# ``data.features`` / ``data.free_runtimes`` never KeyErrors on the error path. -# ``features`` is populated with the canonical FREE_FEATURES set so a caller -# that lands on this branch does not silently see an empty feature list -# (which would look like "OSS install has no free features -- lock everything" -# once enforcement is live). -_MINIMAL_OSS_FREE_SNAPSHOT = { - # Resolver unavailable => plan unknown. Never let a paywall read this - # snapshot as a confirmed free plan (see entitlements.plan_pending()). - "pending": True, - "tier": "oss", - "tier_label": "OSS", - "tier_rank": 0, - "source": "oss", - "node_limit": 1, - "channel_limit": None, - "expiry": None, - "expired": False, - "days_until_expiry": None, - "is_paid": False, - "grace": True, - "enforced": False, - "enforce_at": None, - "enforce_at_iso": None, - "days_until_enforce": None, - "retention_days": 7, - "effective_retention_days": 7, - "runtimes": ["nemoclaw", "openclaw"], - "features": sorted( - [ - "brain", - "channels", - "crons", - "flow", - "health", - "logs", - "nemo_governance", - "overview", - "sessions", - "tracing", - "transcripts", - "usage", - ] - ), - "free_runtimes": ["nemoclaw", "openclaw"], - "paid_runtimes": [], - "all_runtimes": ["nemoclaw", "openclaw"], - "locked_runtimes": [], - "locked_features": [], - "next_tier": None, - "next_tier_label": None, - "prev_tier": None, - "prev_tier_label": None, - "next_tier_diff": None, - "prev_tier_diff": None, - "next_tier_capacity_diff": None, - "prev_tier_capacity_diff": None, - "next_tier_unlocks": None, - "prev_tier_unlocks": None, - "next_tier_locks": None, - "prev_tier_locks": None, - # Parity with Entitlement.to_dict(). The fallback branch below merges - # the *live* value on top of these keys so the overlay never sees a - # stale ``False`` on the resolver-crashed path (which would let an - # expired-trial user silently through). - "hard_blocked": False, - "free_only_mode": False, -} - - - -def _ingest_is_running() -> bool: - """True when something is actually writing the local store. - - The first-run panel must not tell a user to "run some work through the - agent" when the real reason they see nothing is that nothing is reading - it. `clawmetry` alone starts the dashboard only: every - `_start_daemon_background()` call site sits in the cloud-connect flow, so - a plain `pip install clawmetry && clawmetry` never ingests (#5740). - - Best effort, and it fails toward silence: an error answers True so the - panel says nothing about ingest rather than accusing a healthy install. - """ - try: - from clawmetry import local_store as _ls - if getattr(_ls, "_writer_owner", False): - return True - if _ls._daemon_registered(): - return True - return _ls.DB_PATH.exists() - except Exception: - return True - - -@bp_entitlement.route("/api/entitlement") -def api_entitlement(): - try: - from clawmetry import entitlements as _ent - - out = _ent.get_entitlement().to_dict() - # "We do not know this account's plan yet" — distinct from "this - # account is on the free plan". Gating UI must not lock while true. - out["pending"] = _ent.plan_pending() - return jsonify(out) - except Exception as exc: - logger.warning( - "api_entitlement: primary resolver failed, falling back to OSS-free: %s", - exc, - ) - # Preferred fallback: build the canonical OSS-free entitlement and return - # its ``to_dict()`` so the shape matches the healthy path exactly (30+ - # keys including ``features``, ``free_runtimes``, ``channel_limit``, - # ``next_tier_capacity_diff``, ``next_tier_locks``...). Only if this ALSO - # fails -- typically an import failure of the entitlements module itself - # -- do we fall through to the in-line snapshot below. - try: - from clawmetry import entitlements as _ent - - degraded = _ent._oss_free().to_dict() - # The resolver just failed, so this is "unknown", never "confirmed - # free". Gating UI must not lock on it. - degraded["pending"] = True - return jsonify(degraded) - except Exception as exc2: - logger.warning( - "api_entitlement: OSS-free fallback also failed, using minimal snapshot: %s", - exc2, - ) - snap = dict(_MINIMAL_OSS_FREE_SNAPSHOT) - # Merge the live hard-block signal on top of the frozen snapshot so the - # overlay never sees a stale ``False`` on the resolver-crashed path. - try: - from clawmetry import trial_enforcement as _te - snap["hard_blocked"] = bool(_te.is_hard_blocked()) - snap["free_only_mode"] = bool(_te.free_only_mode_enabled()) - except Exception: - pass - return jsonify(snap) - - -@bp_entitlement.route("/api/entitlement/refresh", methods=["POST"]) -def api_entitlement_refresh(): - try: - from clawmetry import entitlements as _ent - - _ent.invalidate() - return jsonify(_ent.get_entitlement(force=True).to_dict()) - except Exception as exc: - logger.warning("api_entitlement_refresh: falling back to OSS-free: %s", exc) - return jsonify( - { - "tier": "oss", - "tier_label": "OSS", - "tier_rank": 0, - "source": "oss", - "node_limit": 1, - "expiry": None, - "expired": False, - "is_paid": False, - "grace": True, - "enforced": False, - "enforce_at": None, - "enforce_at_iso": None, - "days_until_enforce": None, - "runtimes": ["nemoclaw", "openclaw"], - "features": [], - "locked_runtimes": [], - "locked_features": [], - "next_tier_diff": None, - "prev_tier_diff": None, - "next_tier_unlocks": None, - "prev_tier_unlocks": None, - } - ) - - -@bp_entitlement.route("/api/entitlement/upgrade-diff") -def api_entitlement_upgrade_diff(): - try: - target = (request.args.get("target") or "").strip().lower() - from clawmetry import entitlements as _ent - - return jsonify(_ent.upgrade_diff(target)) - except Exception as exc: - logger.warning("api_entitlement_upgrade_diff: error: %s", exc) - return jsonify( - { - "target": (request.args.get("target") or "").strip().lower(), - "added_features": [], - "added_runtimes": [], - } - ) - - -@bp_entitlement.route("/api/entitlement/downgrade-diff") -def api_entitlement_downgrade_diff(): - try: - target = (request.args.get("target") or "").strip().lower() - from clawmetry import entitlements as _ent - - return jsonify(_ent.downgrade_diff(target)) - except Exception as exc: - logger.warning("api_entitlement_downgrade_diff: error: %s", exc) - return jsonify( - { - "target": (request.args.get("target") or "").strip().lower(), - "lost_features": [], - "lost_runtimes": [], - } - ) - - -@bp_entitlement.route("/api/entitlement/tier-diff") -def api_entitlement_tier_diff(): - """``GET /api/entitlement/tier-diff?from=&to=`` -- arbitrary- - endpoint diff between any two tiers, generalising ``/upgrade-diff`` / - ``/downgrade-diff`` (which pin one endpoint to the resolved entitlement) - to ANY pair so a "Compare A vs B" pricing-page widget can render the - transition between any two rungs without first switching the resolver. - - The payload carries both ``added_*`` and ``lost_*`` lists on every call, - plus a ``direction`` tag (``upgrade`` | ``downgrade`` | ``lateral`` | - ``identity``) and a ``capacity_changes`` dict for the three capacity - axes (channels / retention / nodes), so the same shape covers all four - transition kinds and the consumer reads the tag instead of inferring - direction from the deltas. - - ``400`` when ``from=`` or ``to=`` is missing; ``404`` when either id - is unknown. ``trial`` IS accepted -- it is unreachable via the - purchasable-only helpers but is a valid hypothetical endpoint. - Never 5xxs: a resolver failure short-circuits to ``404`` instead of - raising so a paywall surface keeps rendering. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - body = _ent.tier_diff(f, t) - if body is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - return jsonify(body) - except Exception as exc: - logger.warning("api_entitlement_tier_diff: error: %s", exc) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/tier-path") -def api_entitlement_tier_path(): - """``GET /api/entitlement/tier-path?from=&to=`` -- arbitrary- - endpoint stepwise path between any two tiers; the path analogue of - ``/api/entitlement/tier-diff``, generalising ``/upgrade-path`` / - ``/downgrade-path`` (which pin one endpoint to the resolved - entitlement) to ANY pair so a "Compare A vs B" pricing-page widget - can render the rung sequence between any two tiers without first - switching the resolver. - - Each row in ``path`` is a full :func:`clawmetry.entitlements.tier_diff` - payload between the previous step in the path (or ``from`` for the - first row) and the current rung -- so each row is a marginal step - diff. Same-rank siblings strictly between the endpoints are both - included; same-rank siblings of the destination are excluded so the - path terminates exactly at ``to``. - - Response shape:: - - { - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "path": [, ...], - } - - Identity (``from == to``) returns an empty path. Lateral (same rank, - different id) returns a single-row path. ``400`` when ``from=`` or - ``to=`` is missing; ``404`` when either id is unknown. ``trial`` IS - accepted as an endpoint -- it is excluded from the walked rungs (not - purchasable) but the endpoint computation still resolves. Never - 5xxs: a resolver failure short-circuits to ``404`` so a pricing-page - surface keeps rendering instead of breaking. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - path = _ent.tier_path(f, t) - if path is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - from_rank = _ent.tier_rank(f) - to_rank = _ent.tier_rank(t) - if f == t: - direction = "identity" - elif from_rank == to_rank: - direction = "lateral" - elif to_rank > from_rank: - direction = "upgrade" - else: - direction = "downgrade" - return jsonify( - { - "from": f, - "from_label": _ent.tier_label(f), - "from_rank": from_rank, - "to": t, - "to_label": _ent.tier_label(t), - "to_rank": to_rank, - "direction": direction, - "path": path, - } - ) - except Exception as exc: - logger.warning("api_entitlement_tier_path: error: %s", exc) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/tier-diff-batch") -def api_entitlement_tier_diff_batch(): - """``GET /api/entitlement/tier-diff-batch`` -- full marginal - :func:`tier_diff` for every purchasable tier in one pass. Plural - sibling of ``/api/entitlement/tier-diff`` and the "all-slices-in-one- - row" member of the batch family alongside ``/tier-unlocks-batch`` - (feature/runtime grant slice), ``/tier-locks-batch`` (feature/ - runtime loss slice) and ``/capacity-diff-batch`` (capacity slice). - Where each of those siblings carries a single slice of the per-rung - transition, this endpoint carries ALL slices (``added_features`` + - ``lost_features`` + ``added_runtimes`` + ``lost_runtimes`` + - ``capacity_changes``) in one row so a pricing-page UI can render the - full marginal column off **one** round-trip instead of N calls to - ``/tier-diff``. - - Anchor matches ``/tier-unlocks-batch``: each row is the - :func:`clawmetry.entitlements.tier_diff` payload between the next- - lower-rank purchasable tier and the current rung. At the floor - (``TIER_OSS`` / ``TIER_CLOUD_FREE``) the row collapses to an - identity diff (``from == to``, ``direction == "identity"``, empty - marginal lists) -- every row stays byte-stable with a valid - ``/tier-diff`` payload so the singular and batch never diverge in - shape. - - Response shape:: - - { - "tiers": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/tier-diff`` exactly - (``from``, ``from_label``, ``from_rank``, ``to``, ``to_label``, - ``to_rank``, ``direction``, ``added_features``, ``lost_features``, - ``added_runtimes``, ``lost_runtimes``, ``capacity_changes``). The - trial tier is excluded -- it is not purchasable, same posture as the - other batches. Never 5xxs: a resolver failure yields an empty - ``tiers`` list and the grace-shape envelope so the pricing page - keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - rows = _ent.tier_diff_batch() - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_tier_diff_batch: error: %s", exc) - return jsonify( - { - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/capacity-diff") -def api_entitlement_capacity_diff(): - """``GET /api/entitlement/capacity-diff?target=`` -- per-axis - capacity transition (channels / retention / nodes) from the resolved - entitlement to ``target``. Companion to ``/upgrade-diff`` (feature + - runtime adds) and ``/downgrade-diff`` (feature + runtime losses). The - payload is direction-agnostic: each axis carries the same - ``{before, after, delta, unlocked, locked}`` triple so both the - upgrade-to and the cancellation-to CTAs read off one shape.""" - try: - target = (request.args.get("target") or "").strip().lower() - from clawmetry import entitlements as _ent - - return jsonify(_ent.capacity_diff(target)) - except Exception as exc: - logger.warning("api_entitlement_capacity_diff: error: %s", exc) - return jsonify( - { - "target": (request.args.get("target") or "").strip().lower(), - "channel_limit": None, - "retention_days": None, - "node_limit": None, - } - ) - - -@bp_entitlement.route("/api/entitlement/capacity-headroom") -def api_entitlement_capacity_headroom(): - """``GET /api/entitlement/capacity-headroom?channels=&retention_days=&nodes=`` - -- per-axis "how much room is left" against the resolved entitlement's - capacity caps. - - Resolver-pinned companion to - ``/api/entitlement/tiers-for-capacity-batch`` (which is decoupled from - the resolver and returns the full pricing ladder): given caller- - supplied *current usage* on any of the three capacity axes, returns one - row per supplied axis describing how close to (or past) the current - tier's cap the caller is. A quota gauge or a "you're at 4/5 channels - on Starter" badge reads off this single primitive without re-deriving - per-tier caps client-side. - - Envelope shape:: - - { - "tier": "", - "tier_label": "", - "channels": | None, - "retention_days": | None, - "nodes": | None, - } - - Each ```` (matches :func:`clawmetry.entitlements._headroom_row` - byte-for-byte):: - - { - "kind": "channels" | "retention_days" | "nodes", - "used": , - "cap": | None, - "remaining": | None, - "is_unlimited": , - "at_limit": , - "over_limit": , - "pct_used": | None, - } - - Per-axis ``None`` means "axis not supplied" (matches - ``/tiers-for-capacity-batch``'s "None means unset, not unlimited" - posture). A blank, non-int, negative, or ``bool``-in-disguise - (``?channels=true``) value on any axis short-circuits that axis to - ``None`` -- a stray query string cannot silently blank a gauge; the - caller opts in per-axis by supplying a real int. - - In grace mode :meth:`Entitlement.channel_limit` returns ``None`` - (unlimited), so every axis collapses to the unlimited-side row shape - and the gauge renders "unlimited / N used" instead of a bogus - percentage while the grace window is still open. - - Never 5xxs: on a resolver failure the neutral envelope shape - (``tier=oss``, every axis ``None``) is returned so a paywall tile - keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - kwargs: dict[str, int] = {} - for name in ("channels", "retention_days", "nodes"): - present, ok, val, _raw = _parse_capacity_arg(name) - if present and ok and val is not None and val >= 0: - kwargs[name] = val - return jsonify(_ent.capacity_headroom(**kwargs)) - except Exception as exc: - logger.warning("api_entitlement_capacity_headroom: error: %s", exc) - try: - from clawmetry import entitlements as _ent - - return jsonify( - { - "tier": _ent.TIER_OSS, - "tier_label": _ent.tier_label(_ent.TIER_OSS), - "channels": None, - "retention_days": None, - "nodes": None, - } - ) - except Exception: - return jsonify( - { - "tier": "oss", - "tier_label": "OSS", - "channels": None, - "retention_days": None, - "nodes": None, - } - ) - - -@bp_entitlement.route("/api/entitlement/capacity-headroom-at") -def api_entitlement_capacity_headroom_at(): - """``GET /api/entitlement/capacity-headroom-at?tier=&channels=&retention_days=&nodes=`` - -- hypothetical-perspective sibling of - ``/api/entitlement/capacity-headroom``: per-axis headroom against a - caller-supplied ``tier``'s static caps rather than the resolved - entitlement. - - Fills the ``_at`` slot on the capacity-headroom axis alongside - ``/tiers-for-channel-count-at`` / ``/tiers-for-retention-window-at`` - / ``/tiers-for-node-count-at``, so a pricing-page "what would my - usage look like on tier X?" walk-through can call every ``_at`` - endpoint uniformly. - - Response shape mirrors ``/api/entitlement/capacity-headroom`` - byte-for-byte; the ``tier`` echo carries the perspective. Returns - ``{"error": "unknown tier"}`` + HTTP 404 for an empty / unknown - ``?tier=`` (matches ``/tier-diff-at`` / ``/tiers-for-batch-at``). - - Decoupled from the resolved entitlement (walks the static per-tier - caps), so grace vs enforce yields byte-identical rows. Never 5xxs. - """ - try: - from clawmetry import entitlements as _ent - - perspective = (request.args.get("tier") or "").strip().lower() - if not perspective: - return jsonify({"error": "unknown tier"}), 404 - kwargs: dict[str, int] = {} - for name in ("channels", "retention_days", "nodes"): - present, ok, val, _raw = _parse_capacity_arg(name) - if present and ok and val is not None and val >= 0: - kwargs[name] = val - row = _ent.capacity_headroom_at(perspective, **kwargs) - if row is None: - return jsonify({"error": "unknown tier"}), 404 - return jsonify(row) - except Exception as exc: - logger.warning("api_entitlement_capacity_headroom_at: error: %s", exc) - return jsonify({"error": "capacity-headroom-at failed"}), 500 - - -def _neighbour_tier_headroom_envelope( - *, direction: str, headroom: dict | None -) -> dict: - """Shared envelope for ``/next-tier-capacity-headroom`` + - ``/previous-tier-capacity-headroom``. - - Wraps the raw :func:`clawmetry.entitlements.capacity_headroom_at` row - in the same "current-tier context + null-at-boundary" shape as - ``/next-tier-unlocks`` / ``/previous-tier-unlocks`` (see - ``api_entitlement_next_tier_unlocks``) so an upgrade / downgrade card - can bind against ``headroom`` as the payload with the boundary case - surfacing as ``headroom=null`` at HTTP 200 -- callers never have to - branch on status code. - """ - from clawmetry import entitlements as _ent - - try: - ent = _ent.get_entitlement() - return { - "current_tier": ent.tier, - "current_tier_label": _ent.tier_label(ent.tier), - "current_tier_rank": _ent.tier_rank(ent.tier), - "direction": direction, - "headroom": headroom, - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - except Exception: - return { - "current_tier": "oss", - "current_tier_label": "OSS", - "current_tier_rank": 0, - "direction": direction, - "headroom": None, - "grace": True, - "enforced": False, - } - - -@bp_entitlement.route("/api/entitlement/next-tier-capacity-headroom") -def api_entitlement_next_tier_capacity_headroom(): - """``GET /api/entitlement/next-tier-capacity-headroom?channels= - &retention_days=&nodes=`` -- per-axis headroom envelope for - the tier immediately above the resolved entitlement, given the caller- - supplied per-axis usage. - - Scalar "one rung up" sibling of ``/api/entitlement/capacity-headroom``. - Composes ``next_purchasable_tier()`` + ``capacity_headroom_at(next)`` - so an upgrade-CTA card can render "here's what your gauges would look - like on " off ONE call instead of a resolve + at-tier - round-trip. Sits alongside the caps-only - ``/api/entitlement/next-tier-capacity-diff`` (which reports cap - deltas without folding in current usage) and the marginal-features - ``/api/entitlement/next-tier-unlocks``. - - Response shape (matches ``/api/entitlement/next-tier-unlocks``'s - envelope byte-for-key so an upgrade-CTA can bind the two off one - fetch shape):: - - { - "current_tier": "", - "current_tier_label": "", - "current_tier_rank": , - "direction": "upgrade", - "headroom": | null, - "grace": , - "enforced": , - } - - Each ```` -- when non-null -- matches - ``/api/entitlement/capacity-headroom-at`` byte-for-byte (``tier`` - echoing the next-tier id, per-axis rows in the - :func:`entitlements._headroom_row` shape) so an existing - ``/capacity-headroom-at`` renderer consumes the ``headroom`` field - unchanged. - - ``headroom`` is ``null`` (still HTTP 200) when the resolved - entitlement is already on the top rung (no next-purchasable tier), - so the CTA can hide itself off ``headroom == null`` instead of - branching on status code. Same per-axis "None means axis not - supplied" posture as ``/api/entitlement/capacity-headroom`` -- an - axis the caller didn't pass stays ``None`` on the inner row. Same - bad-arg short-circuit as ``/capacity-headroom`` (blank / non-int / - negative axis stays ``None``). - - Decoupled from grace vs enforce on the headroom side -- - ``capacity_headroom_at`` walks the static per-tier caps, so the - inner rows are byte-identical across modes. The "next tier" - identity itself still tracks the live resolver, so an operator - moving from ``cloud_starter`` to ``cloud_pro`` sees the target - flip once activation lands. - - Never 5xxs: on a resolver / delegation failure returns the neutral - grace envelope (``current_tier=oss``, ``headroom=null``) so a - paywall tile keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - kwargs: dict[str, int] = {} - for name in ("channels", "retention_days", "nodes"): - present, ok, val, _raw = _parse_capacity_arg(name) - if present and ok and val is not None and val >= 0: - kwargs[name] = val - row = _ent.next_tier_capacity_headroom(**kwargs) - return jsonify( - _neighbour_tier_headroom_envelope( - direction="upgrade", headroom=row - ) - ) - except Exception as exc: - logger.warning( - "api_entitlement_next_tier_capacity_headroom: error: %s", exc - ) - return jsonify( - _neighbour_tier_headroom_envelope( - direction="upgrade", headroom=None - ) - ) - - -@bp_entitlement.route("/api/entitlement/previous-tier-capacity-headroom") -def api_entitlement_previous_tier_capacity_headroom(): - """``GET /api/entitlement/previous-tier-capacity-headroom?channels= - &retention_days=&nodes=`` -- per-axis headroom envelope for - the tier immediately below the resolved entitlement, given the caller- - supplied per-axis usage. - - Downgrade twin of ``/api/entitlement/next-tier-capacity-headroom``. - Composes ``previous_purchasable_tier()`` + - ``capacity_headroom_at(prev)`` so a downgrade-preview card can show - "here's what would break on " -- axes whose inner - ``over_limit`` flips ``True`` are exactly the ones the caller would - lose headroom on. Sits alongside the caps-only - ``/api/entitlement/previous-tier-capacity-diff`` and the marginal- - features ``/api/entitlement/previous-tier-unlocks``. - - Envelope shape matches - ``/api/entitlement/next-tier-capacity-headroom`` byte-for-key -- - with ``direction`` echoing ``"downgrade"`` -- so a single renderer - can consume both. Inner ``headroom`` (when non-null) matches - ``/api/entitlement/capacity-headroom-at`` byte-for-byte. - - ``headroom`` is ``null`` (still HTTP 200) when the resolved - entitlement is already on the bottom rung (no previous-purchasable - tier) so the downgrade card can hide itself off ``headroom == null`` - instead of branching on status code. Same per-axis "None means axis - not supplied" posture, bad-arg short-circuit, and grace / enforce - invariance as ``/api/entitlement/next-tier-capacity-headroom``. - Never 5xxs. - """ - try: - from clawmetry import entitlements as _ent - - kwargs: dict[str, int] = {} - for name in ("channels", "retention_days", "nodes"): - present, ok, val, _raw = _parse_capacity_arg(name) - if present and ok and val is not None and val >= 0: - kwargs[name] = val - row = _ent.previous_tier_capacity_headroom(**kwargs) - return jsonify( - _neighbour_tier_headroom_envelope( - direction="downgrade", headroom=row - ) - ) - except Exception as exc: - logger.warning( - "api_entitlement_previous_tier_capacity_headroom: error: %s", exc - ) - return jsonify( - _neighbour_tier_headroom_envelope( - direction="downgrade", headroom=None - ) - ) - - -@bp_entitlement.route("/api/entitlement/capacity-headroom-batch") -def api_entitlement_capacity_headroom_batch(): - """``GET /api/entitlement/capacity-headroom-batch?channels= - &retention_days=&nodes=`` -- per-tier headroom envelope for - every purchasable tier in one pass, given the caller-supplied per-axis - usage. - - Plural sibling of ``/api/entitlement/capacity-headroom-at``: where the - singular ``_at`` endpoint returns one hypothetical tier's per-axis - envelope, the batch returns the same envelope for every entry in - :data:`entitlements._PURCHASABLE_TIERS` so a pricing-page "at each - tier, would my usage fit?" table can render every rung off **one** - round-trip instead of N calls to ``/capacity-headroom-at``. - - Fills the ``_batch`` slot on the capacity-headroom axis alongside - ``/capacity-diff-batch`` (per-tier per-axis transition triples against - the resolved entitlement) and the per-axis - ``/tiers-for-channel-count-batch`` / ``/tiers-for-retention-window-batch`` - / ``/tiers-for-node-count-batch`` families. - - Response shape:: - - { - "tiers": [, ...], - "current_tier": "", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/capacity-headroom-at`` for - the same axis inputs byte-for-byte - (``tier`` / ``tier_label`` / ``channels`` / ``retention_days`` / - ``nodes``, with each per-axis row matching the - :func:`entitlements._headroom_row` shape). Rows are sorted by tier - rank ascending with ``id`` as a stable tiebreaker -- byte-stable - against ``/capacity-diff-batch`` / ``/tier-unlocks-batch`` / - ``/tier-locks-batch`` / ``/preview-batch`` so a pricing table lines - up rung-for-rung without client-side re-sort. The trial tier is - excluded (mirrors the other ``*-batch`` siblings -- not purchasable). - - Per-axis ``None`` on every row means "axis not supplied" (matches - ``/capacity-headroom`` and ``/tiers-for-capacity-batch``'s posture). - A blank, non-int, negative, or ``bool``-in-disguise value on any - axis short-circuits that axis to ``None`` on every row -- a stray - query string cannot silently blank the whole ladder. - - Decoupled from the resolved entitlement -- every row walks the - static per-tier caps -- so grace vs enforce yields byte-identical - ``tiers`` payloads. The envelope's ``current_tier`` / ``grace`` / - ``enforced`` still track the live resolver so the UI can highlight - the caller's current rung on the ladder. - - Never 5xxs: on a resolver failure the empty-tiers grace envelope is - returned so a pricing table falls back to an empty list instead of - 500-ing. - """ - try: - from clawmetry import entitlements as _ent - - kwargs: dict[str, int] = {} - for name in ("channels", "retention_days", "nodes"): - present, ok, val, _raw = _parse_capacity_arg(name) - if present and ok and val is not None and val >= 0: - kwargs[name] = val - rows = _ent.capacity_headroom_batch(**kwargs) - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning( - "api_entitlement_capacity_headroom_batch: error: %s", exc - ) - return jsonify( - { - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/capacity-headroom-path") -def api_entitlement_capacity_headroom_path(): - """``GET /api/entitlement/capacity-headroom-path?from=&to= - &channels=&retention_days=&nodes=`` -- per-rung - capacity-headroom envelope along an arbitrary ``from -> to`` segment, - given the caller-supplied per-axis usage. - - Path analogue of ``/capacity-headroom-batch`` (which walks every - purchasable tier) and headroom-shaped mirror of - ``/capacity-diff-path`` (per-rung marginal capacity transitions), - ``/tier-unlocks-path`` (marginal grants per rung), - ``/tier-locks-path`` (marginal losses per rung) and - ``/preview-path`` (cumulative ``Entitlement.to_dict`` per rung) -- - the fifth member of the capacity axis's ``_path`` family. Lets an - upgrade-walkthrough surface render the "watch your headroom recover - rung by rung" view off ONE round-trip without re-deriving per-tier - caps in JS. - - Rung walk matches ``/capacity-diff-path`` / ``/preview-path`` / - ``/tier-unlocks-path`` / ``/tier-locks-path`` byte-for-byte (same - ``_PURCHASABLE_TIERS`` filter + same ``(rank, id)`` / - ``(-rank, id)`` sort key + same destination-sibling exclusion), so - the rung ids from this endpoint line up rung-for-rung with those - four siblings. Same-rank siblings between the endpoints are both - included; same-rank siblings of the destination are excluded so the - path terminates exactly at ``to``. - - Response shape:: - - { - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "path": [, ...], - } - - Each ```` matches ``/api/entitlement/capacity-headroom-at`` for - the same axis inputs byte-for-byte (``tier`` / ``tier_label`` / - ``channels`` / ``retention_days`` / ``nodes``, with each per-axis row - matching the :func:`entitlements._headroom_row` shape). Identity - (``from == to``) returns an empty ``path``. Lateral (same rank, - different id) returns a single-row path. ``400`` when ``from=`` or - ``to=`` is missing; ``404`` when either id is unknown. ``trial`` IS - accepted as an endpoint -- excluded from the walked rungs (not - purchasable) but the lateral / identity branch still resolves. - - Per-axis ``None`` on every row means "axis not supplied" (matches - ``/capacity-headroom-batch``'s posture). A blank, non-int, negative, - or ``bool``-in-disguise value on any axis short-circuits that axis - to ``None`` on every row -- a stray query string cannot silently - blank the whole walk. - - Decoupled from the resolved entitlement -- every rung walks the - static per-tier caps via - :func:`entitlements.capacity_headroom_at` -- so grace vs enforce - yields byte-identical ``path`` payloads. Never 5xxs: a resolver - failure short-circuits to ``404`` so an upgrade-walkthrough surface - keeps rendering instead of breaking. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - kwargs: dict[str, int] = {} - for name in ("channels", "retention_days", "nodes"): - present, ok, val, _raw = _parse_capacity_arg(name) - if present and ok and val is not None and val >= 0: - kwargs[name] = val - path = _ent.capacity_headroom_path(f, t, **kwargs) - if path is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - from_rank = _ent.tier_rank(f) - to_rank = _ent.tier_rank(t) - if f == t: - direction = "identity" - elif from_rank == to_rank: - direction = "lateral" - elif to_rank > from_rank: - direction = "upgrade" - else: - direction = "downgrade" - return jsonify( - { - "from": f, - "from_label": _ent.tier_label(f), - "from_rank": from_rank, - "to": t, - "to_label": _ent.tier_label(t), - "to_rank": to_rank, - "direction": direction, - "path": path, - } - ) - except Exception as exc: - logger.warning( - "api_entitlement_capacity_headroom_path: error: %s", exc - ) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/capacity-diff-batch") -def api_entitlement_capacity_diff_batch(): - """``GET /api/entitlement/capacity-diff-batch`` -- per-axis capacity - transition for every purchasable tier in one pass. Plural sibling of - ``/api/entitlement/capacity-diff``: where the singular endpoint - returns one tier's per-axis triple, the batch returns the full - pricing-page ladder in tier-rank order so a pricing-table UI can - render the capacity column ("channels: 3 -> unlimited, retention: - 7d -> 30d, nodes: 1 -> unlimited") off **one** round-trip instead - of N calls. - - Direction-agnostic capacity companion to ``/tier-unlocks-batch`` - (marginal feature / runtime grant per rung), ``/tier-locks-batch`` - (marginal feature / runtime loss per rung) and ``/preview-batch`` - (cumulative ``Entitlement.to_dict`` shape per rung): pair them to - render the full "what's at X / what's new at X / what you'd give - up at X / capacity at X" view of a pricing table without - client-side composition. - - Response shape:: - - { - "tiers": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/capacity-diff`` exactly - (``target``, ``channel_limit``, ``retention_days``, ``node_limit`` - where each axis is the same ``{before, after, delta, unlocked, - locked}`` triple). The trial tier is excluded -- not purchasable, - same posture as the other ``*-batch`` siblings. Never 5xxs: a - resolver failure yields an empty ``tiers`` list and the grace-shape - envelope so the pricing page keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - rows = _ent.capacity_diff_batch() - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_capacity_diff_batch: error: %s", exc) - return jsonify( - { - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/capacity-diff-at") -def api_entitlement_capacity_diff_at(): - """``GET /api/entitlement/capacity-diff-at?tier=&target=`` - -- scalar what-if sibling of ``/api/entitlement/capacity-diff``: per- - axis capacity transition (channels / retention / nodes) from a - caller-supplied ``tier`` to ``target``, computed off the static - per-tier caps rather than the resolved entitlement - ``/capacity-diff`` anchors to. - - Lets a pricing-comparison tooltip render "capacity at B vs A" for - any ``(A, B)`` pair in one round-trip without fetching the full - ``/capacity-diff-path?from=A&to=B`` payload and reading the - destination row client-side. The returned row matches the - destination row of ``/capacity-diff-path`` for the same pair -- - a parity test pins this so the scalar what-if and the path-walker - cannot drift. - - Accepts any tier id in :data:`entitlements._TIER_FEATURES` on either - argument (including ``trial``), matching the other ``_at`` family - endpoints. Direction is not normalised: an upgrade pair flips - ``unlocked`` on axes that go from a finite cap to unlimited; a - downgrade pair flips ``locked`` on axes that go from unlimited to - finite; identity / lateral-rank pairs collapse every axis to a - no-op triple. - - Response shape:: - - { - "tier": "", - "target": "", - "row": {}, - } - - The inner ``row`` matches the singular ``/capacity-diff`` row shape - exactly (``target``, ``channel_limit``, ``retention_days``, - ``node_limit`` where each axis is the same ``{before, after, delta, - unlocked, locked}`` triple) -- with the ``before`` side carrying the - caller-supplied ``tier``'s static caps (NOT the resolved - entitlement's caps the singular endpoint uses). - - - **400** when either ``tier=`` or ``target=`` is missing / blank. - - **404** when ``tier`` or ``target`` is unknown. The body carries - ``which`` so a caller can render the right "unknown ..." message. - - **Never 5xxs**: builder failure falls through to a 404 so the - tooltip surface stays mute instead of breaking. - """ - raw_tier = request.args.get("tier") - tier_in = (raw_tier or "").strip().lower() - if not tier_in: - return jsonify({"error": "missing tier"}), 400 - raw_target = request.args.get("target") - target_in = (raw_target or "").strip().lower() - if not target_in: - return jsonify({"error": "missing target"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier_in not in _ent._TIER_FEATURES: - return ( - jsonify( - {"error": "unknown tier", "which": "tier", "tier": tier_in} - ), - 404, - ) - if target_in not in _ent._TIER_FEATURES: - return ( - jsonify( - { - "error": "unknown target", - "which": "target", - "target": target_in, - } - ), - 404, - ) - row = _ent.capacity_diff_at(tier_in, target_in) - if row is None: - return ( - jsonify( - { - "error": "capacity-diff-at failed", - "tier": tier_in, - "target": target_in, - } - ), - 404, - ) - return jsonify({"tier": tier_in, "target": target_in, "row": row}) - except Exception as exc: - logger.warning("api_entitlement_capacity_diff_at: error: %s", exc) - return ( - jsonify( - { - "error": "capacity-diff-at failed", - "tier": tier_in, - "target": target_in, - } - ), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/capacity-diff-at-batch") -def api_entitlement_capacity_diff_at_batch(): - """``GET /api/entitlement/capacity-diff-at-batch?tier=`` -- - what-if + batch sibling of ``/api/entitlement/capacity-diff-batch``: - per-axis capacity-transition rows for every purchasable tier as a - target, computed against the caller-supplied ``tier`` rather than - the resolved entitlement ``/capacity-diff-batch`` anchors to. - - Composes the scalar what-if (``/capacity-diff-at``) and the live - batch (``/capacity-diff-batch``) -- same row shape and ordering as - the live batch, same hypothetical perspective as the ``_at`` - endpoint. Lets a pricing-comparison matrix UI render the "capacity - vs " column for every rung off **one** round- - trip instead of N calls to ``/capacity-diff-at``. - - Pair with ``/tier-unlocks-at-batch`` (marginal feature/runtime - grant per rung) and ``/tier-locks-at-batch`` (marginal loss per - rung) to render the full "what's new at X / what you'd give up at - X / capacity at X" view of a pricing matrix pivoted around any - hypothetical perspective tier without client-side composition. - - Accepts any tier id in :data:`entitlements._TIER_FEATURES` on the - ``tier`` arg (including ``trial``), matching the other ``_at`` - family endpoints. The target list mirrors ``/capacity-diff-batch`` - (purchasable tiers only -- trial excluded), so the rows match the - live batch's target axis byte-for-byte and the response can be - folded into the same pricing-page table. - - Response shape:: - - { - "tier": "", - "tiers": [, ...], - "current_tier": "", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/capacity-diff-at`` for - the same ``(tier, target)`` pair exactly (``target``, - ``channel_limit``, ``retention_days``, ``node_limit``) -- with the - ``before`` side carrying the caller-supplied ``tier``'s static - caps (NOT the resolved entitlement's caps the live batch uses). - - - **400** when ``tier=`` is missing / blank. - - **404** when ``tier`` is unknown. The body carries ``which=tier`` - so a caller can render the right "unknown tier" message. - - **Never 5xxs**: a resolver failure yields an empty ``tiers`` list - and the grace-shape envelope so the matrix keeps rendering. - """ - raw_tier = request.args.get("tier") - tier_in = (raw_tier or "").strip().lower() - if not tier_in: - return jsonify({"error": "missing tier"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier_in not in _ent._TIER_FEATURES: - return ( - jsonify( - {"error": "unknown tier", "which": "tier", "tier": tier_in} - ), - 404, - ) - rows = _ent.capacity_diff_at_batch(tier_in) or [] - ent = _ent.get_entitlement() - return jsonify( - { - "tier": tier_in, - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning( - "api_entitlement_capacity_diff_at_batch: error: %s", exc - ) - return jsonify( - { - "tier": tier_in, - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/tier-diff-at") -def api_entitlement_tier_diff_at(): - """``GET /api/entitlement/tier-diff-at?tier=&from= - &to=`` -- arbitrary-endpoint diff between two tiers, rendered - from a hypothetical ``perspective_tier``. - - What-if sibling of ``/tier-diff``: same payload shape, plus a - ``perspective_tier`` echo so a pricing-comparison tooltip surface - can call ``X_at(perspective, from, to)`` uniformly across the whole - ``_at`` scalar family (alongside ``/capacity-diff-at``, - ``/tier-unlocks-at``, ``/tier-locks-at``, ``/tier-catalog-at`` and - the ``_at_path`` walk siblings). Closes the ``_at`` slot of the - ``tier_diff`` family alongside the existing ``/tier-diff-at-batch`` - (walk every purchasable target from one source) and the - ``/tier-path-at`` walk-shape sibling in the open ``tier_path_at`` - PR. - - Body posture matches ``/tier-catalog-at-path``: perspective is - validated against :data:`_TIER_ORDER` (including :data:`TIER_TRIAL`) - but does NOT shape rows -- the diff is anchored to ``from`` / - ``to``. A parity test pins the response body against - ``/tier-diff?from=&to=`` for every valid perspective so - the ``_at`` prefix cannot silently drift into shaping rows. - - Response body extends the ``/tier-diff`` shape with three extra - fields at the top so a consumer can echo the perspective in a - "Comparing A vs B from perspective P" tooltip without a second - round-trip:: - - { - "perspective_tier": "", - "perspective_tier_rank": , - "perspective_tier_label":"...", - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "added_features": [...], - "lost_features": [...], - "added_runtimes": [...], - "lost_runtimes": [...], - "capacity_changes": {...}, - } - - - **400** when ``tier=``, ``from=`` or ``to=`` is missing / blank. - - **404** when any id is unknown (body carries ``which: "tier" | - "from" | "to"`` so the caller can point at the offender). - - **200** on the happy path with the shape above. - - Never 5xxs: a resolver failure short-circuits to 404 so a - pricing-comparison tooltip keeps rendering instead of breaking. - """ - p = (request.args.get("tier") or "").strip().lower() - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not p: - return jsonify({"error": "missing tier"}), 400 - if not f: - return jsonify({"error": "missing from"}), 400 - if not t: - return jsonify({"error": "missing to"}), 400 - try: - from clawmetry import entitlements as _ent - - if p not in _ent._TIER_ORDER: - return ( - jsonify( - {"error": "unknown tier", "which": "tier", "tier": p} - ), - 404, - ) - if f not in _ent._TIER_FEATURES: - return ( - jsonify( - {"error": "unknown tier", "which": "from", "from": f} - ), - 404, - ) - if t not in _ent._TIER_FEATURES: - return ( - jsonify( - {"error": "unknown tier", "which": "to", "to": t} - ), - 404, - ) - body = _ent.tier_diff_at(p, f, t) - if body is None: - return ( - jsonify( - { - "error": "unknown tier", - "tier": p, - "from": f, - "to": t, - } - ), - 404, - ) - out = { - "perspective_tier": p, - "perspective_tier_rank": _ent.tier_rank(p), - "perspective_tier_label": _ent.tier_label(p), - } - out.update(body) - return jsonify(out) - except Exception as exc: - logger.warning("api_entitlement_tier_diff_at: error: %s", exc) - return ( - jsonify( - { - "error": "unknown tier", - "tier": p, - "from": f, - "to": t, - } - ), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/tier-diff-at-batch") -def api_entitlement_tier_diff_at_batch(): - """``GET /api/entitlement/tier-diff-at-batch?tier=`` -- - what-if + batch sibling of ``/api/entitlement/tier-diff-batch``: - full marginal :func:`tier_diff` payload between the caller-supplied - ``tier`` and every purchasable tier as a target, in one pass. - - Composes the arbitrary-endpoint diff (``/tier-diff``) and the live - batch (``/tier-diff-batch``) -- same row shape and ordering as the - live batch, but every row's ``from`` side is anchored to the - caller-supplied ``tier`` instead of the per-rung next-lower- - purchasable anchor ``/tier-diff-batch`` carries. Lets a pricing- - comparison matrix UI render the "full marginal vs " column for every rung off **one** round-trip instead of N - calls to ``/tier-diff``. - - The "all-slices-in-one-row" member of the ``_at`` batch family - alongside ``/tier-unlocks-at-batch`` (marginal feature/runtime - grant slice), ``/tier-locks-at-batch`` (marginal feature/runtime - loss slice) and ``/capacity-diff-at-batch`` (capacity slice). Pair - them to render the full "what's new at X / what you'd give up at - X / capacity at X" view of a pricing matrix pivoted around any - hypothetical perspective tier without client-side composition; - this endpoint folds the three slices into one row for callers that - prefer a single call. - - Accepts any tier id in :data:`entitlements._TIER_FEATURES` on the - ``tier`` arg (including ``trial``), matching the other ``_at`` - family endpoints. The target list mirrors ``/tier-diff-batch`` - (purchasable tiers only -- trial excluded), so the rows match the - live batch's target axis byte-for-byte and the response can be - folded into the same pricing-page table. - - Response shape:: - - { - "tier": "", - "tiers": [, ...], - "current_tier": "", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/tier-diff`` for the same - ``(from=tier, to=target)`` pair exactly -- ``from``, ``from_label``, - ``from_rank``, ``to``, ``to_label``, ``to_rank``, ``direction``, - ``added_features``, ``lost_features``, ``added_runtimes``, - ``lost_runtimes``, ``capacity_changes`` -- with ``from`` byte-equal - to the caller-supplied ``tier`` on every row. - - - **400** when ``tier=`` is missing / blank. - - **404** when ``tier`` is unknown. The body carries ``which=tier`` - so a caller can render the right "unknown tier" message. - - **Never 5xxs**: a resolver failure yields an empty ``tiers`` list - and the grace-shape envelope so the matrix keeps rendering. - """ - raw_tier = request.args.get("tier") - tier_in = (raw_tier or "").strip().lower() - if not tier_in: - return jsonify({"error": "missing tier"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier_in not in _ent._TIER_FEATURES: - return ( - jsonify( - {"error": "unknown tier", "which": "tier", "tier": tier_in} - ), - 404, - ) - rows = _ent.tier_diff_at_batch(tier_in) or [] - ent = _ent.get_entitlement() - return jsonify( - { - "tier": tier_in, - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning( - "api_entitlement_tier_diff_at_batch: error: %s", exc - ) - return jsonify( - { - "tier": tier_in, - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/capacity-diff-path") -def api_entitlement_capacity_diff_path(): - """``GET /api/entitlement/capacity-diff-path?from=&to=`` -- - per-rung capacity transition along an arbitrary ``from -> to`` segment. - Path analogue of ``/capacity-diff-batch`` (which walks every purchasable - tier off the resolved entitlement); capacity-only mirror of - ``/tier-path`` (which carries the full ``tier_diff`` per rung). Lets a - capacity-only pricing widget render the channels / retention / nodes - marginal steps between any two tiers off ONE round-trip without paying - for the feature / runtime set diff on every row. - - Rung walk matches ``/tier-path``: visit every purchasable tier strictly - between ``from`` and ``to`` plus the destination ``to`` itself, in - tier-rank order. Same-rank siblings between the endpoints are both - included; same-rank siblings of the destination are excluded so the - path terminates exactly at ``to``. Each row's ``before`` side comes - off the previous step's static caps (or ``from`` for the first row), - so a consumer can fold the rows to reconstruct the cumulative - ``tier_diff(from, to)['capacity_changes']`` shape. - - Response shape:: - - { - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "path": [, ...], - } - - Each ```` matches ``/capacity-diff`` exactly - (``target``, ``channel_limit``, ``retention_days``, ``node_limit`` - where each axis is the same ``{before, after, delta, unlocked, - locked}`` triple). Identity (``from == to``) returns an empty path. - Lateral (same rank, different id) returns a single-row path. ``400`` - when ``from=`` or ``to=`` is missing; ``404`` when either id is - unknown. ``trial`` IS accepted as an endpoint -- it is excluded from - the walked rungs (not purchasable) but the endpoint computation - still resolves. Never 5xxs: a resolver failure short-circuits to - ``404`` so a pricing-page surface keeps rendering instead of breaking. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - path = _ent.capacity_diff_path(f, t) - if path is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - from_rank = _ent.tier_rank(f) - to_rank = _ent.tier_rank(t) - if f == t: - direction = "identity" - elif from_rank == to_rank: - direction = "lateral" - elif to_rank > from_rank: - direction = "upgrade" - else: - direction = "downgrade" - return jsonify( - { - "from": f, - "from_label": _ent.tier_label(f), - "from_rank": from_rank, - "to": t, - "to_label": _ent.tier_label(t), - "to_rank": to_rank, - "direction": direction, - "path": path, - } - ) - except Exception as exc: - logger.warning("api_entitlement_capacity_diff_path: error: %s", exc) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/preview") -def api_entitlement_preview(): - """``GET /api/entitlement/preview?tier=`` -- the full - :meth:`Entitlement.to_dict` shape rendered for a hypothetical tier so an - upgrade-CTA card can show concrete numbers ("365-day retention, unlimited - channels, claude_code unlocked") without the client re-deriving per-tier - capacity. ``404`` when the tier id is unknown.""" - target = (request.args.get("tier") or "").strip().lower() - if not target: - return jsonify({"error": "missing tier"}), 400 - try: - from clawmetry import entitlements as _ent - - body = _ent.preview(target) - if body is None: - return jsonify({"error": "unknown tier", "tier": target}), 404 - return jsonify(body) - except Exception as exc: - logger.warning("api_entitlement_preview: error: %s", exc) - return jsonify({"error": "preview failed", "tier": target}), 500 - - -@bp_entitlement.route("/api/entitlement/preview-batch") -def api_entitlement_preview_batch(): - """``GET /api/entitlement/preview-batch`` -- the full - :meth:`Entitlement.to_dict` shape rendered for every purchasable tier - in one pass. Plural sibling of ``/api/entitlement/preview``: where the - singular endpoint returns one tier's row (and 404s on an unknown id), - the batch returns the full pricing-page ladder in tier-rank order so a - pricing-table UI can render the cumulative-state column off **one** - round-trip instead of N calls. - - Cumulative-state companion to ``/api/entitlement/tier-unlocks-batch`` - (marginal grant per rung) and ``/api/entitlement/tier-locks-batch`` - (marginal loss per rung): pair the three to render the "what's at X / - what's new at X / what you'd give up at X" three-column view of a - pricing table without client-side composition. - - Response shape:: - - { - "tiers": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/preview`` exactly -- the - full ``Entitlement.to_dict`` shape with ``source="preview"`` and - ``grace=False`` so concrete per-tier capacity surfaces. The trial - tier is excluded -- it is not purchasable, same posture as the - singular helper. Row order matches ``/api/entitlement/tier-unlocks-batch`` - and ``/api/entitlement/tier-locks-batch`` rung-for-rung. Never 5xxs: - a resolver failure yields an empty ``tiers`` list and the grace-shape - envelope so the pricing page keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - rows = _ent.preview_batch() - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_preview_batch: error: %s", exc) - return jsonify( - { - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/preview-at") -def api_entitlement_preview_at(): - """``GET /api/entitlement/preview-at?tier=&target=`` -- - what-if sibling of ``/api/entitlement/preview``: the full - :meth:`Entitlement.to_dict` snapshot at ``target`` rendered from the - perspective of a hypothetical ``tier``. - - Fills the ``_at`` slot in the preview family alongside - ``/api/entitlement/tier-spec-at``, - ``/api/entitlement/feature-spec-at``, - ``/api/entitlement/runtime-spec-at``, - ``/api/entitlement/capacity-diff-at``, - ``/api/entitlement/tier-unlocks-at``, - ``/api/entitlement/tier-locks-at`` and - ``/api/entitlement/lock-reason-at``. Lets a pricing-comparison - tooltip hydrate one cumulative-state row from a hypothetical - perspective in ONE round-trip using the uniform - ``X_at(perspective, target)`` request shape the rest of the ``_at`` - family already exposes. - - Unlike ``/api/entitlement/preview`` (which 404s the non-purchasable - :data:`TIER_TRIAL`), ``/preview-at`` accepts trial as a target and - returns the trial preview row -- lenient ``_at`` posture matching - ``/tier-spec-at`` / ``/feature-spec-at`` / ``/runtime-spec-at``. The - perspective tier is validated but does not shape the returned row - (byte-parity with :func:`entitlements._preview_row` holds for every - perspective / target combination). - - Row shape matches ``/api/entitlement/preview`` exactly -- the full - ``Entitlement.to_dict`` payload with ``source="preview"`` and - ``grace=False`` so concrete per-tier capacity surfaces. - - - **400** when ``tier=`` or ``target=`` is missing / blank - - **404** when ``tier`` is not in :data:`entitlements._TIER_ORDER` - or ``target`` is not in :data:`entitlements._TIER_FEATURES`; the - body carries ``which`` so a caller can render the right - "unknown ..." message - - **Never 5xxs**: the helper reads only the static per-tier maps, so - a resolver failure short-circuits to ``404`` instead of 500 - """ - raw_tier = request.args.get("tier") - tier = (raw_tier or "").strip().lower() - if not tier: - return jsonify({"error": "missing tier"}), 400 - raw_target = request.args.get("target") - target = (raw_target or "").strip().lower() - if not target: - return jsonify({"error": "missing target"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier not in _ent._TIER_ORDER: - return ( - jsonify({"error": "unknown tier", "which": "tier", "tier": tier}), - 404, - ) - if target not in _ent._TIER_FEATURES: - return ( - jsonify( - { - "error": "unknown target", - "which": "target", - "target": target, - } - ), - 404, - ) - body = _ent.preview_at(tier, target) - if body is None: - return ( - jsonify( - { - "error": "preview-at failed", - "tier": tier, - "target": target, - } - ), - 404, - ) - return jsonify({"tier": tier, "target": target, "preview": body}) - except Exception as exc: - logger.warning("api_entitlement_preview_at: error: %s", exc) - return jsonify({"error": "preview-at failed"}), 500 - - -@bp_entitlement.route("/api/entitlement/preview-at-batch") -def api_entitlement_preview_at_batch(): - """``GET /api/entitlement/preview-at-batch?tier= - &targets=a,b,c`` -- what-if + batch sibling of - ``/api/entitlement/preview-at``. - - Where ``/preview-at`` hydrates ONE cumulative-state row from a - hypothetical perspective, this hydrates N rows for a caller-supplied - subset of target tiers off a single round-trip. Fixed-perspective - multi-target companion of ``/preview-at`` and - caller-supplied-targets sibling of ``/preview-batch`` (which walks - :data:`_PURCHASABLE_TIERS` unconditionally). Fills the ``_at_batch`` - slot alongside ``/api/entitlement/tier-spec-at-batch``, - ``/api/entitlement/feature-spec-at-batch``, - ``/api/entitlement/runtime-spec-at-batch``. - - Use case: a pricing-comparison matrix UI ("from my perspective tier, - render the cumulative-state row for OSS, Cloud Starter, Cloud Pro - and Enterprise") hydrates every column off ONE call instead of N - calls to ``/preview-at``. - - Each ``tiers[]`` entry is byte-identical to a row from - :func:`entitlements.preview_at` (and therefore - :func:`entitlements._preview_row`) for the same target -- pinned by - the parity tests so the scalar / batch what-if accessors cannot - drift. Supplied ids are normalised (whitespace stripped, lowercased, - duplicates dropped, first-seen order preserved). Unknown ids do not - 404 the call -- they are echoed in ``unknown[]`` so a partially-bad - caller still gets rows back for the valid ids alongside a list of - what was dropped. - - Response shape (mirrors ``/tier-spec-at-batch`` / - ``/feature-spec-at-batch`` / ``/runtime-spec-at-batch`` plus a - ``perspective_tier`` echo):: - - { - "tiers": [, ...], - "unknown": ["bogus_id", ...], - "perspective_tier": "...", - "perspective_tier_rank": , - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - - **400** when ``tier=`` is missing / blank or ``targets=`` is - missing / empty after normalisation - - **404** when ``tier`` is unknown (body carries ``which: "tier"``) - - **Never 5xxs**: a resolver failure short-circuits to the OSS-free - shape (empty rows, ``current_tier=oss``, ``grace=true``) with the - perspective tier echoed so the UI keeps rendering. - """ - raw_tier = request.args.get("tier") - tier_in = (raw_tier or "").strip().lower() - if not tier_in: - return jsonify({"error": "missing tier"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier_in not in _ent._TIER_ORDER: - return ( - jsonify( - {"error": "unknown tier", "which": "tier", "tier": tier_in} - ), - 404, - ) - targets = _parse_csv_arg("targets") - if not targets: - return ( - jsonify({"error": "supply targets="}), - 400, - ) - batch = _ent.preview_at_batch(tier_in, targets) - if batch is None: - batch = {"tiers": [], "unknown": []} - ent = _ent.get_entitlement() - batch["perspective_tier"] = tier_in - batch["perspective_tier_rank"] = _ent.tier_rank(tier_in) - batch["current_tier"] = ent.tier - batch["current_tier_rank"] = _ent.tier_rank(ent.tier) - batch["grace"] = bool(ent.grace) - batch["enforced"] = _ent.is_enforced() - return jsonify(batch) - except Exception as exc: - logger.warning( - "api_entitlement_preview_at_batch: error: %s", exc - ) - return jsonify( - { - "tiers": [], - "unknown": [], - "perspective_tier": tier_in, - "perspective_tier_rank": 0, - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/preview-path") -def api_entitlement_preview_path(): - """``GET /api/entitlement/preview-path?from=&to=`` -- - arbitrary-endpoint stepwise cumulative-state path between any two - tiers; the cumulative-state analogue of ``/tier-path`` (full - ``tier_diff`` per rung), ``/capacity-diff-path`` (capacity-only per - rung), ``/tier-unlocks-path`` (marginal grants per rung) and - ``/tier-locks-path`` (marginal losses per rung) -- the fifth and - final member of the ``_path`` family, the path-shaped sibling of - ``/preview-batch``. Lets an upgrade-walkthrough surface render the - "Cloud Pro: 90-day retention, unlimited channels, claude_code - unlocked" card at every rung between any two tiers off ONE - round-trip, without re-deriving capacity in JS. - - Each row in ``path`` is the full - :meth:`Entitlement.to_dict` payload at that rung -- identical shape - to a single ``/preview`` row, with ``source="preview"`` and - ``grace=False`` so concrete per-tier capacity surfaces. Rung walk - is byte-stable against ``/tier-path``, ``/capacity-diff-path``, - ``/tier-unlocks-path`` and ``/tier-locks-path`` (same - ``_PURCHASABLE_TIERS`` filter + same sort + same destination-sibling - exclusion), so the five paths line up rung-for-rung. - - Response shape:: - - { - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "path": [, ...], - } - - Direction semantics: - - * ``upgrade`` (ascending) -- rows climb cumulatively rung by rung. - * ``downgrade`` (descending) -- rows shrink cumulatively rung by - rung; the cancellation-walkthrough counterpart. - * ``lateral`` (same rank, different id) -- single-row path; row - carries the cumulative preview at ``to``. - * ``identity`` (``from == to``) -- empty path; no rungs to walk. - - Same-rank siblings strictly between the endpoints are both - included; same-rank siblings of the destination are excluded so the - path terminates exactly at ``to``. ``400`` when ``from=`` or ``to=`` - is missing; ``404`` when either id is unknown. ``trial`` IS accepted - as an endpoint -- it is excluded from the walked intermediate rungs - (not purchasable) but is a valid endpoint via the lateral branch. - Never 5xxs: a resolver failure short-circuits to ``404`` so an - upgrade-walkthrough surface keeps rendering instead of breaking. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - path = _ent.preview_path(f, t) - if path is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - from_rank = _ent.tier_rank(f) - to_rank = _ent.tier_rank(t) - if f == t: - direction = "identity" - elif from_rank == to_rank: - direction = "lateral" - elif to_rank > from_rank: - direction = "upgrade" - else: - direction = "downgrade" - return jsonify( - { - "from": f, - "from_label": _ent.tier_label(f), - "from_rank": from_rank, - "to": t, - "to_label": _ent.tier_label(t), - "to_rank": to_rank, - "direction": direction, - "path": path, - } - ) - except Exception as exc: - logger.warning("api_entitlement_preview_path: error: %s", exc) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/tier-unlocks") -def api_entitlement_tier_unlocks(): - """``GET /api/entitlement/tier-unlocks?tier=`` -- marginal unlocks - for ``tier`` (features + runtimes that first become available at that - tier vs the next-lower purchasable tier). Sibling of ``/preview`` - (cumulative shape). ``404`` when the tier id is unknown (including - ``trial`` -- not purchasable).""" - target = (request.args.get("tier") or "").strip().lower() - if not target: - return jsonify({"error": "missing tier"}), 400 - try: - from clawmetry import entitlements as _ent - - body = _ent.tier_unlocks(target) - if body is None: - return jsonify({"error": "unknown tier", "tier": target}), 404 - return jsonify(body) - except Exception as exc: - logger.warning("api_entitlement_tier_unlocks: error: %s", exc) - return jsonify({"error": "tier-unlocks failed", "tier": target}), 500 - - -@bp_entitlement.route("/api/entitlement/tier-unlocks-batch") -def api_entitlement_tier_unlocks_batch(): - """``GET /api/entitlement/tier-unlocks-batch`` -- marginal unlocks for - every purchasable tier in one pass. Plural sibling of - ``/api/entitlement/tier-unlocks``: where the singular endpoint - returns one tier's row (and 404s on an unknown id), the batch - returns the full pricing-page ladder in tier-rank order so a - pricing-table UI can render the "what's new in X" column off - **one** round-trip instead of N calls. - - Response shape:: - - { - "tiers": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/tier-unlocks`` exactly - (``tier``, ``tier_label``, ``tier_rank``, ``previous_tier``, - ``previous_tier_label``, ``previous_tier_rank``, ``features``, - ``runtimes``). The trial tier is excluded -- it is not purchasable, - same posture as the singular helper. Never 5xxs: a resolver failure - yields an empty ``tiers`` list and the grace-shape envelope so the - pricing page keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - rows = _ent.tier_unlocks_batch() - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_tier_unlocks_batch: error: %s", exc) - return jsonify( - { - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/tier-unlocks-path") -def api_entitlement_tier_unlocks_path(): - """``GET /api/entitlement/tier-unlocks-path?from=&to=`` -- - arbitrary-endpoint stepwise unlock path between any two tiers; the - unlocks-focused analogue of ``/tier-path`` (full ``tier_diff`` per - rung) and ``/capacity-diff-path`` (capacity-only per rung). Lets an - upgrade-walkthrough surface render only the *newly-unlocked* features - + runtimes at each rung between any two tiers off ONE round-trip, - without the noise of the capacity axes or the symmetric ``lost_*`` - lists ``/tier-path`` carries. - - Each row in ``path`` is a :func:`clawmetry.entitlements.tier_unlocks` - payload between the previous step in the path (or ``from`` for the - first row) and the current rung -- so each row is a marginal-step - unlock and a consumer can fold ``features`` / ``runtimes`` across - rows to reconstruct the cumulative - ``tier_diff(from, to)['added_*']`` shape (the same chain-property - ``/tier-path`` and ``/capacity-diff-path`` enforce on their rows). - Same-rank siblings strictly between the endpoints are both included; - same-rank siblings of the destination are excluded so the path - terminates exactly at ``to``. Rung walk is byte-stable against - ``/tier-path`` and ``/capacity-diff-path``. - - Response shape:: - - { - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "path": [, ...], - } - - Each ```` matches the singular ``/tier-unlocks`` row shape - exactly (``tier``, ``tier_label``, ``tier_rank``, ``previous_tier``, - ``previous_tier_label``, ``previous_tier_rank``, ``features``, - ``runtimes``) -- with ``previous_tier`` chained from the path (the - previous step), NOT the global next-lower-purchasable-tier anchor - the singular helper uses. - - Direction semantics: - - * ``upgrade`` (ascending) -- each row's ``features`` / ``runtimes`` - are the marginal grant at that rung. - * ``downgrade`` (descending) -- each row's ``features`` / - ``runtimes`` are typically empty (use ``/tier-path`` for the - marginal-loss view of a downgrade). The path still walks rungs so - a UI keyed off rung shape keeps working. - * ``lateral`` (same rank, different id) -- single-row path; carries - the set difference between the two same-rank tier grants. - * ``identity`` (``from == to``) -- empty path; no rungs to walk. - - Identity (``from == to``) returns an empty path. Lateral (same rank, - different id) returns a single-row path. ``400`` when ``from=`` or - ``to=`` is missing; ``404`` when either id is unknown. ``trial`` IS - accepted as an endpoint -- it is excluded from the walked rungs (not - purchasable) but the endpoint computation still resolves. Never - 5xxs: a resolver failure short-circuits to ``404`` so an upgrade- - walkthrough surface keeps rendering instead of breaking. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - path = _ent.tier_unlocks_path(f, t) - if path is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - from_rank = _ent.tier_rank(f) - to_rank = _ent.tier_rank(t) - if f == t: - direction = "identity" - elif from_rank == to_rank: - direction = "lateral" - elif to_rank > from_rank: - direction = "upgrade" - else: - direction = "downgrade" - return jsonify( - { - "from": f, - "from_label": _ent.tier_label(f), - "from_rank": from_rank, - "to": t, - "to_label": _ent.tier_label(t), - "to_rank": to_rank, - "direction": direction, - "path": path, - } - ) - except Exception as exc: - logger.warning("api_entitlement_tier_unlocks_path: error: %s", exc) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/next-tier-unlocks") -def api_entitlement_next_tier_unlocks(): - """``GET /api/entitlement/next-tier-unlocks`` -- marginal unlocks row - for the rung immediately above the resolved entitlement, in - :func:`clawmetry.entitlements.tier_unlocks` shape (``tier``, - ``tier_label``, ``tier_rank``, ``previous_tier``, ``previous_tier_label``, - ``previous_tier_rank``, ``features``, ``runtimes``). - - Current-relative convenience for ``/api/entitlement/tier-unlocks - ?tier=``; the upgrade-CTA companion to - ``/api/entitlement/next-tier-diff`` (same marginal, ``upgrade_diff`` - shape). Returns ``{"unlocks": null, ...}`` at the ceiling - (no rung above to upgrade to). Never 5xxs: a resolver failure - short-circuits to the grace-shape envelope so the dashboard CTA - keeps rendering instead of disappearing. - """ - try: - from clawmetry import entitlements as _ent - - ent = _ent.get_entitlement() - body = ent.next_tier_unlocks() - return jsonify( - { - "current_tier": ent.tier, - "current_tier_label": _ent.tier_label(ent.tier), - "current_tier_rank": _ent.tier_rank(ent.tier), - "unlocks": body, - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_next_tier_unlocks: error: %s", exc) - return jsonify( - { - "current_tier": "oss", - "current_tier_label": "OSS", - "current_tier_rank": 0, - "unlocks": None, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/previous-tier-unlocks") -def api_entitlement_previous_tier_unlocks(): - """``GET /api/entitlement/previous-tier-unlocks`` -- marginal unlocks row - for the rung immediately below the resolved entitlement, in - :func:`clawmetry.entitlements.tier_unlocks` shape. - - Current-relative convenience for ``/api/entitlement/tier-unlocks - ?tier=``. Useful as a downgrade-confirmation - detail row alongside :func:`previous_tier_diff` -- ``features`` / - ``runtimes`` here are what the rung below *first* unlocked vs the rung - below it (a tier-property), so a "you'd still keep X" copy can - reference the same set the rung-below was originally sold on. Returns - ``{"unlocks": null, ...}`` at the floor (no rung below). Never 5xxs. - """ - try: - from clawmetry import entitlements as _ent - - ent = _ent.get_entitlement() - body = ent.previous_tier_unlocks() - return jsonify( - { - "current_tier": ent.tier, - "current_tier_label": _ent.tier_label(ent.tier), - "current_tier_rank": _ent.tier_rank(ent.tier), - "unlocks": body, - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_previous_tier_unlocks: error: %s", exc) - return jsonify( - { - "current_tier": "oss", - "current_tier_label": "OSS", - "current_tier_rank": 0, - "unlocks": None, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/next-tier-locks") -def api_entitlement_next_tier_locks(): - """``GET /api/entitlement/next-tier-locks`` -- marginal locks row for the - rung immediately above the resolved entitlement, in - :func:`clawmetry.entitlements.tier_locks` shape (``tier``, - ``tier_label``, ``tier_rank``, ``next_tier``, ``next_tier_label``, - ``next_tier_rank``, ``lost_features``, ``lost_runtimes``). - - Symmetric companion to ``/api/entitlement/next-tier-unlocks``: that - endpoint carries the rung-above's first-grant row, this carries its - first-loss row -- a pricing-table cell can render both off ONE - entitlement round-trip. ``locks`` is ``null`` at the ladder's - ceiling (no rung above). Never 5xxs: a resolver failure - short-circuits to the grace-shape envelope so the dashboard CTA - keeps rendering instead of disappearing. - """ - try: - from clawmetry import entitlements as _ent - - ent = _ent.get_entitlement() - body = ent.next_tier_locks() - return jsonify( - { - "current_tier": ent.tier, - "current_tier_label": _ent.tier_label(ent.tier), - "current_tier_rank": _ent.tier_rank(ent.tier), - "locks": body, - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_next_tier_locks: error: %s", exc) - return jsonify( - { - "current_tier": "oss", - "current_tier_label": "OSS", - "current_tier_rank": 0, - "locks": None, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/previous-tier-locks") -def api_entitlement_previous_tier_locks(): - """``GET /api/entitlement/previous-tier-locks`` -- marginal locks row for - the rung immediately below the resolved entitlement, in - :func:`clawmetry.entitlements.tier_locks` shape. - - The step-down confirmation detail row paired with - ``/api/entitlement/previous-tier-diff`` (which carries the same - marginal in ``downgrade_diff`` shape). ``lost_features`` / - ``lost_runtimes`` here are what the rung below first loses vs the - rung above it -- and since "the rung above" the previous purchasable - tier *is* the caller's current tier in the simple single-step - downgrade case, these lists byte-equal the caller's marginal loss - when stepping down by one rung. ``locks`` is ``null`` at the - ladder's floor (no rung below). Never 5xxs. - """ - try: - from clawmetry import entitlements as _ent - - ent = _ent.get_entitlement() - body = ent.previous_tier_locks() - return jsonify( - { - "current_tier": ent.tier, - "current_tier_label": _ent.tier_label(ent.tier), - "current_tier_rank": _ent.tier_rank(ent.tier), - "locks": body, - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_previous_tier_locks: error: %s", exc) - return jsonify( - { - "current_tier": "oss", - "current_tier_label": "OSS", - "current_tier_rank": 0, - "locks": None, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/tier-locks") -def api_entitlement_tier_locks(): - """``GET /api/entitlement/tier-locks?tier=`` -- marginal locks for - ``tier`` (features + runtimes that disappear when descending from - the next-higher purchasable tier into ``tier``). Marginal-loss - companion to ``/tier-unlocks``: where the unlocks endpoint answers - "what does X first unlock vs the tier below it", this answers "what - does X first lose vs the tier above it" -- the per-rung - downgrade-warning row a step-down CTA renders, paired with - ``/downgrade-path`` the way ``/tier-unlocks`` is paired with - ``/upgrade-path``. - - Returns ``404`` when the tier id is unknown (including ``trial`` -- - not purchasable). Enterprise callers get a populated envelope with - ``next_tier=null`` and empty loss lists (nothing above to step down - from), not a 404 -- the tier is valid, the marginal just collapses - to nothing. - """ - target = (request.args.get("tier") or "").strip().lower() - if not target: - return jsonify({"error": "missing tier"}), 400 - try: - from clawmetry import entitlements as _ent - - body = _ent.tier_locks(target) - if body is None: - return jsonify({"error": "unknown tier", "tier": target}), 404 - return jsonify(body) - except Exception as exc: - logger.warning("api_entitlement_tier_locks: error: %s", exc) - return jsonify({"error": "tier-locks failed", "tier": target}), 500 - - -@bp_entitlement.route("/api/entitlement/tier-locks-batch") -def api_entitlement_tier_locks_batch(): - """``GET /api/entitlement/tier-locks-batch`` -- marginal locks for - every purchasable tier in one pass. Plural sibling of - ``/api/entitlement/tier-locks``: where the singular endpoint - returns one tier's row (and 404s on an unknown id), the batch - returns the full purchasable ladder in tier-rank order so a - downgrade-warning matrix can render the "what you'd give up at X" - column off **one** round-trip instead of N calls. - - Marginal-loss companion to ``/api/entitlement/tier-unlocks-batch``: - pair the two endpoints to render the upgrade-CTA + downgrade-warning - columns on a pricing table without client-side composition. - - Response shape:: - - { - "tiers": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/tier-locks`` exactly - (``tier``, ``tier_label``, ``tier_rank``, ``next_tier``, - ``next_tier_label``, ``next_tier_rank``, ``lost_features``, - ``lost_runtimes``). The trial tier is excluded -- it is not - purchasable, same posture as the singular helper. Never 5xxs: a - resolver failure yields an empty ``tiers`` list and the grace-shape - envelope so the downgrade-warning UI keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - rows = _ent.tier_locks_batch() - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": rows, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_tier_locks_batch: error: %s", exc) - return jsonify( - { - "tiers": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/tier-locks-path") -def api_entitlement_tier_locks_path(): - """``GET /api/entitlement/tier-locks-path?from=&to=`` -- - arbitrary-endpoint stepwise marginal-loss path between any two - tiers; the locks-focused mirror of ``/tier-unlocks-path`` and the - fourth member of the ``_path`` family alongside ``/tier-path`` (full - ``tier_diff`` per rung) and ``/capacity-diff-path`` (capacity-only - per rung). Lets a downgrade-walkthrough surface render only the - *newly-lost* features + runtimes at each rung between any two tiers - off ONE round-trip, without the noise of the capacity axes or the - symmetric ``added_*`` lists ``/tier-path`` carries. - - Each row in ``path`` is a :func:`clawmetry.entitlements.tier_locks` - payload between the previous step in the path (or ``from`` for the - first row) and the current rung -- so each row is a marginal-step - loss and a consumer can fold ``lost_features`` / ``lost_runtimes`` - across rows to reconstruct the cumulative - ``tier_diff(from, to)['lost_*']`` shape (the same chain-property - ``/tier-path``, ``/capacity-diff-path``, and ``/tier-unlocks-path`` - enforce on their rows). Same-rank siblings strictly between the - endpoints are both included; same-rank siblings of the destination - are excluded so the path terminates exactly at ``to``. Rung walk is - byte-stable against ``/tier-path``, ``/capacity-diff-path``, and - ``/tier-unlocks-path``. - - Response shape:: - - { - "from": "", - "from_label": "...", - "from_rank": , - "to": "", - "to_label": "...", - "to_rank": , - "direction": "upgrade" | "downgrade" | "lateral" | "identity", - "path": [, ...], - } - - Each ```` matches the singular ``/tier-locks`` row shape - exactly (``tier``, ``tier_label``, ``tier_rank``, ``next_tier``, - ``next_tier_label``, ``next_tier_rank``, ``lost_features``, - ``lost_runtimes``) -- with ``next_tier`` chained from the path (the - previous step), NOT the global next-higher-purchasable-tier anchor - the singular helper uses. - - Direction semantics: - - * ``downgrade`` (descending) -- each row's ``lost_features`` / - ``lost_runtimes`` are the marginal loss at that rung. - * ``upgrade`` (ascending) -- each row's ``lost_features`` / - ``lost_runtimes`` are typically empty (use ``/tier-unlocks-path`` - for the marginal-grant view of an upgrade). The path still walks - rungs so a UI keyed off rung shape keeps working. - * ``lateral`` (same rank, different id) -- single-row path; carries - the set difference (``from`` minus ``to``) between the two - same-rank tier grants. - * ``identity`` (``from == to``) -- empty path; no rungs to walk. - - Identity (``from == to``) returns an empty path. Lateral (same rank, - different id) returns a single-row path. ``400`` when ``from=`` or - ``to=`` is missing; ``404`` when either id is unknown. ``trial`` IS - accepted as an endpoint -- it is excluded from the walked rungs (not - purchasable) but the endpoint computation still resolves. Never - 5xxs: a resolver failure short-circuits to ``404`` so a downgrade- - walkthrough surface keeps rendering instead of breaking. - """ - f = (request.args.get("from") or "").strip().lower() - t = (request.args.get("to") or "").strip().lower() - if not f or not t: - return jsonify({"error": "missing from or to"}), 400 - try: - from clawmetry import entitlements as _ent - - path = _ent.tier_locks_path(f, t) - if path is None: - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - from_rank = _ent.tier_rank(f) - to_rank = _ent.tier_rank(t) - if f == t: - direction = "identity" - elif from_rank == to_rank: - direction = "lateral" - elif to_rank > from_rank: - direction = "upgrade" - else: - direction = "downgrade" - return jsonify( - { - "from": f, - "from_label": _ent.tier_label(f), - "from_rank": from_rank, - "to": t, - "to_label": _ent.tier_label(t), - "to_rank": to_rank, - "direction": direction, - "path": path, - } - ) - except Exception as exc: - logger.warning("api_entitlement_tier_locks_path: error: %s", exc) - return ( - jsonify({"error": "unknown tier", "from": f, "to": t}), - 404, - ) - - -@bp_entitlement.route("/api/entitlement/upgrade-path") -def api_entitlement_upgrade_path(): - """``GET /api/entitlement/upgrade-path`` -- ordered marginal-unlock - ladder from the resolved tier upward. - - Current-user-relative sibling of ``/api/entitlement/tier-unlocks-batch``: - where the batch returns the full purchasable ladder, this returns only - tiers whose rank is *strictly above* the caller's resolved entitlement - rank, so an upgrade-CTA wizard renders its step sequence without - client-side filtering. - - Response shape:: - - { - "path": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/tier-unlocks`` exactly - (``tier``, ``tier_label``, ``tier_rank``, ``previous_tier``, - ``previous_tier_label``, ``previous_tier_rank``, ``features``, - ``runtimes``). Enterprise callers get an empty ``path`` (already at - the top). Never 5xxs: a resolver failure yields ``path: []`` with the - grace-shape envelope so the upgrade CTA keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - ent = _ent.get_entitlement() - return jsonify( - { - "path": _ent.upgrade_path(), - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_upgrade_path: error: %s", exc) - return jsonify( - { - "path": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/downgrade-path") -def api_entitlement_downgrade_path(): - """``GET /api/entitlement/downgrade-path`` -- ordered cumulative-loss - ladder from the resolved tier downward. - - Direction-flipped sibling of ``/api/entitlement/upgrade-path``: rows - cover the purchasable tiers whose rank is strictly *below* the caller's - resolved entitlement rank, closest rung first. Lets a downgrade-warning - surface render every rung's full loss list without per-tier round-trips. - - Response shape:: - - { - "path": [, ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` carries the destination tier metadata + the caller's - current-tier context + ``lost_features`` / ``lost_runtimes`` cumulative - over the gap (see :func:`clawmetry.entitlements.downgrade_path`). Floor - callers (OSS / Cloud Free) get an empty ``path`` -- no rung below to - descend to. Never 5xxs: a resolver failure yields ``path: []`` with the - grace-shape envelope so the downgrade CTA keeps rendering. - """ - try: - from clawmetry import entitlements as _ent - - ent = _ent.get_entitlement() - return jsonify( - { - "path": _ent.downgrade_path(), - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_downgrade_path: error: %s", exc) - return jsonify( - { - "path": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/upgrade-path-at") -def api_entitlement_upgrade_path_at(): - """``GET /api/entitlement/upgrade-path-at?tier=`` -- scalar - what-if sibling of ``/api/entitlement/upgrade-path``: ordered - marginal-unlock ladder from the caller-supplied ``tier`` upward. - - Source-anchored equivalent of ``/upgrade-path`` (which pins the - walk's starting point to the resolver) -- the ``_at`` sibling in - the ladder-walk family alongside ``/next-tier-spec-at``, - ``/next-tier-unlocks-at``, ``/next-tier-locks-at``, - ``/next-tier-diff-at`` and ``/next-tier-capacity-diff-at``. - Lets a pricing-page "from tier X" wizard render the full upgrade - ladder for any hypothetical source rung without first switching - the resolver. - - Response shape mirrors ``/upgrade-path`` with the ``current_tier`` / - ``current_tier_rank`` echo replaced by the caller-supplied ``tier``:: - - { - "tier": "", - "tier_label": "", - "tier_rank": , - "path": [, ...], - "current_tier": "", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` matches ``/api/entitlement/tier-unlocks`` exactly - (``tier``, ``tier_label``, ``tier_rank``, ``previous_tier``, ...), - byte-identical to the corresponding row in ``/upgrade-path`` when - ``tier`` equals the resolved entitlement -- pinned by parity tests so - the source-anchored and live variants cannot drift. - - Missing / empty ``tier`` -> 400. Unknown ``tier`` -> 404. Enterprise - source -> 200 with ``path: []`` (already at the top). Never 5xxs: a - resolver failure yields the grace-shape envelope so the wizard keeps - rendering. - """ - tier = (request.args.get("tier") or "").strip().lower() - if not tier: - return jsonify({"error": "tier query parameter is required"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier not in _ent._TIER_ORDER: - return jsonify({"error": "unknown tier", "tier": tier}), 404 - path = _ent.upgrade_path_at(tier) or [] - ent = _ent.get_entitlement() - return jsonify( - { - "tier": tier, - "tier_label": _ent.tier_label(tier), - "tier_rank": _ent.tier_rank(tier), - "path": path, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_upgrade_path_at: error: %s", exc) - return jsonify( - { - "tier": tier, - "tier_label": tier, - "tier_rank": -1, - "path": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/downgrade-path-at") -def api_entitlement_downgrade_path_at(): - """``GET /api/entitlement/downgrade-path-at?tier=`` -- scalar - what-if sibling of ``/api/entitlement/downgrade-path``: ordered - cumulative-loss ladder from the caller-supplied ``tier`` downward. - - Source-anchored mirror of ``/api/entitlement/upgrade-path-at`` and - downgrade-side counterpart of the live ``/downgrade-path`` (source - pinned to the resolver). Lets a "compare from tier X" downgrade- - warning surface render every rung's loss list for any hypothetical - source without first asking the resolver. - - Response shape mirrors ``/downgrade-path`` with the ``current_tier`` / - ``current_tier_rank`` echo replaced by the caller-supplied ``tier``:: - - { - "tier": "", - "tier_label": "", - "tier_rank": , - "path": [, ...], - "current_tier": "", - "current_tier_rank": , - "grace": , - "enforced": , - } - - Each ```` carries the destination tier metadata + the walk's - source echo (``current_tier`` / ``current_tier_label`` / - ``current_tier_rank`` retain their :func:`downgrade_path` names for - byte-shape parity, and carry the ``_at`` source in this variant) + - ``lost_features`` / ``lost_runtimes`` cumulative over the gap (see - :func:`clawmetry.entitlements.downgrade_path_at`). - - Missing / empty ``tier`` -> 400. Unknown ``tier`` -> 404. Floor - source (oss / cloud_free) -> 200 with ``path: []`` (no rung strictly - below). Never 5xxs: a resolver failure yields the grace-shape - envelope so the surface keeps rendering. - """ - tier = (request.args.get("tier") or "").strip().lower() - if not tier: - return jsonify({"error": "tier query parameter is required"}), 400 - try: - from clawmetry import entitlements as _ent - - if tier not in _ent._TIER_ORDER: - return jsonify({"error": "unknown tier", "tier": tier}), 404 - path = _ent.downgrade_path_at(tier) or [] - ent = _ent.get_entitlement() - return jsonify( - { - "tier": tier, - "tier_label": _ent.tier_label(tier), - "tier_rank": _ent.tier_rank(tier), - "path": path, - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning("api_entitlement_downgrade_path_at: error: %s", exc) - return jsonify( - { - "tier": tier, - "tier_label": tier, - "tier_rank": -1, - "path": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/upgrade-path-at-batch") -def api_entitlement_upgrade_path_at_batch(): - """``GET /api/entitlement/upgrade-path-at-batch?tiers=a,b,c`` -- batch - what-if sibling of ``/api/entitlement/upgrade-path-at``. - - Where ``/upgrade-path-at`` hydrates the marginal-unlock ladder above - ONE hypothetical source tier, this hydrates it for N hypothetical - sources in ONE round-trip. Pairs with ``/upgrade-path-at`` the same - way ``/tier-catalog-at-batch`` pairs with ``/tier-catalog-at``: - scalar what-if -> matrix what-if across the perspective-tier axis. - - Use case: a pricing-comparison matrix UI ("show me the upgrade - ladder as if I were on OSS vs Cloud Starter vs Cloud Pro vs - Enterprise -- side by side") hydrates every column off ONE call - instead of N calls to ``/upgrade-path-at``. - - Each ``tiers[].path`` list is byte-identical to the body of - ``/upgrade-path-at?tier=`` (its ``path`` field) for the same - source tier -- pinned by parity tests so the scalar and batch - what-if upgrade-path helpers cannot drift. Supplied tier ids are - normalised (whitespace stripped, lowercased, duplicates dropped, - first-seen order preserved). Unknown ids do not 404 the call -- - they are echoed in ``unknown[]`` so a partially-bad caller still - gets rows back for the valid ids alongside a list of what was - dropped, matching every other ``_at_batch`` sibling's posture. - - A source at the ceiling of the purchasable ladder (Enterprise) - still yields a valid row with an empty ``path`` list -- the - ceiling is NOT ``unknown``. Only ids not in :data:`_TIER_ORDER` - (or where the scalar returns ``None``) land in ``unknown[]``. - - Response shape:: - - { - "tiers": [ - { - "tier": "", - "tier_label": "...", - "tier_rank": , - "path": [, ...], - }, - ... - ], - "unknown": ["bogus_id", ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - - **400** when ``tiers=`` is missing / empty after normalisation - - **200** with bucketed unknowns for unknown tier ids -- does NOT - 404 the call, matching every other batch sibling - - **Never 5xxs**: a synthesis failure short-circuits to an envelope - with empty rows so the matrix keeps rendering. - """ - tiers = _parse_csv_arg("tiers") - if not tiers: - return jsonify({"error": "supply tiers="}), 400 - try: - from clawmetry import entitlements as _ent - - batch = _ent.upgrade_path_at_batch(tiers) - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": batch.get("tiers", []), - "unknown": batch.get("unknown", []), - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning( - "api_entitlement_upgrade_path_at_batch: error: %s", exc - ) - return jsonify( - { - "tiers": [], - "unknown": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -@bp_entitlement.route("/api/entitlement/downgrade-path-at-batch") -def api_entitlement_downgrade_path_at_batch(): - """``GET /api/entitlement/downgrade-path-at-batch?tiers=a,b,c`` -- - batch what-if sibling of ``/api/entitlement/downgrade-path-at``. - - Direction-flipped twin of ``/api/entitlement/upgrade-path-at-batch``: - where the upgrade batch hydrates the marginal-unlock ladder strictly - above each source, this hydrates the cumulative-loss ladder strictly - below each source. Same envelope, same per-source row shape, same - unknown-bucketing posture -- only the inner ``path`` list changes - direction. - - Use case: a "compare from tier X" downgrade-warning matrix UI ("show - me the cumulative-loss ladder as if I were on OSS vs Cloud Starter - vs Cloud Pro vs Enterprise -- side by side") hydrates every column - off ONE call instead of N calls to ``/downgrade-path-at``. - - Each ``tiers[].path`` list is byte-identical to the body of - ``/downgrade-path-at?tier=`` (its ``path`` field) for the same - source tier -- pinned by parity tests so the scalar and batch - what-if downgrade-path helpers cannot drift. Supplied tier ids are - normalised (whitespace stripped, lowercased, duplicates dropped, - first-seen order preserved). Unknown ids do not 404 the call -- - they are echoed in ``unknown[]``. - - A source at the floor of the purchasable ladder (``oss`` / - ``cloud_free``) still yields a valid row with an empty ``path`` - list -- the floor is NOT ``unknown``. Only ids not in - :data:`_TIER_ORDER` (or where the scalar returns ``None``) land in - ``unknown[]``. - - Response shape:: - - { - "tiers": [ - { - "tier": "", - "tier_label": "...", - "tier_rank": , - "path": [, ...], - }, - ... - ], - "unknown": ["bogus_id", ...], - "current_tier": "...", - "current_tier_rank": , - "grace": , - "enforced": , - } - - - **400** when ``tiers=`` is missing / empty after normalisation - - **200** with bucketed unknowns for unknown tier ids -- does NOT - 404 the call, matching every other batch sibling - - **Never 5xxs**: a synthesis failure short-circuits to an envelope - with empty rows so the matrix keeps rendering. - """ - tiers = _parse_csv_arg("tiers") - if not tiers: - return jsonify({"error": "supply tiers="}), 400 - try: - from clawmetry import entitlements as _ent - - batch = _ent.downgrade_path_at_batch(tiers) - ent = _ent.get_entitlement() - return jsonify( - { - "tiers": batch.get("tiers", []), - "unknown": batch.get("unknown", []), - "current_tier": ent.tier, - "current_tier_rank": _ent.tier_rank(ent.tier), - "grace": bool(ent.grace), - "enforced": _ent.is_enforced(), - } - ) - except Exception as exc: - logger.warning( - "api_entitlement_downgrade_path_at_batch: error: %s", exc - ) - return jsonify( - { - "tiers": [], - "unknown": [], - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - ) - - -_CAPACITY_PARAMS = ("channels", "retention_days", "nodes") - - -def _parse_capacity_arg(name: str) -> tuple[bool, bool, int | None, str]: - """Parse a capacity query param. - - Returns ``(present, parsed_ok, value, raw)``. ``present`` is True iff the - caller supplied the param at all (even with an empty value, so blank input - doesn't silently fall through to a feature/runtime branch). ``parsed_ok`` - is False when the supplied value couldn't be coerced to ``int`` -- the - HTTP wrapper then short-circuits to ``required_tier=None`` instead of - handing ``None`` to the underlying helper (where, for retention, ``None`` - is the *unlimited* sentinel and would mis-route to Enterprise). - """ - raw = request.args.get(name) - if raw is None: - return False, False, None, "" - raw_stripped = raw.strip() - if not raw_stripped: - return True, False, None, raw_stripped - try: - return True, True, int(raw_stripped), raw_stripped - except (TypeError, ValueError): - return True, False, None, raw_stripped - - -@bp_entitlement.route("/api/entitlement/required-tier") -def api_entitlement_required_tier(): - try: - from clawmetry import entitlements as _ent - - feature = (request.args.get("feature") or "").strip().lower() - runtime = (request.args.get("runtime") or "").strip().lower() - ( - channels_present, - channels_ok, - channels_n, - channels_raw, - ) = _parse_capacity_arg("channels") - ( - retention_present, - retention_ok, - retention_n, - retention_raw, - ) = _parse_capacity_arg("retention_days") - ( - nodes_present, - nodes_ok, - nodes_n, - nodes_raw, - ) = _parse_capacity_arg("nodes") - - supplied = [ - bool(feature), - bool(runtime), - channels_present, - retention_present, - nodes_present, - ] - n_supplied = sum(1 for s in supplied if s) - if n_supplied == 0: - return ( - jsonify( - { - "error": ( - "supply exactly one of feature=, runtime=, " - "channels=, retention_days=, or " - "nodes=" - ) - } - ), - 400, - ) - if n_supplied > 1: - return ( - jsonify( - { - "error": ( - "supply only one of feature=, runtime=, channels=, " - "retention_days=, or nodes=" - ) - } - ), - 400, - ) - - ent = _ent.get_entitlement() - if feature: - key, kind = feature, "feature" - required = _ent.min_tier_for_feature(feature) - allowed = ent.allows_feature(feature) - elif runtime: - key, kind = runtime, "runtime" - required = _ent.min_tier_for_runtime(runtime) - allowed = ent.allows_runtime(runtime) - elif channels_present: - key, kind = channels_raw, "channels" - if channels_ok: - required = _ent.min_tier_for_channel_count(channels_n) - allowed = ent.allows_channel_count(channels_n) - else: - required = None - allowed = True - elif retention_present: - key, kind = retention_raw, "retention_days" - if retention_ok: - required = _ent.min_tier_for_retention_window(retention_n) - allowed = ent.allows_retention_window(retention_n) - else: - required = None - allowed = True - else: - key, kind = nodes_raw, "nodes" - if nodes_ok: - required = _ent.min_tier_for_node_count(nodes_n) - allowed = ent.allows_node_count(nodes_n) - else: - required = None - allowed = True - cur_rank = _ent.tier_rank(ent.tier) - req_rank = _ent.tier_rank(required) if required else -1 - required_label = _ent.tier_label(required) if required else None - return jsonify( - { - "key": key, - "kind": kind, - "required_tier": required, - "required_tier_label": required_label, - "required_tier_rank": req_rank, - "current_tier": ent.tier, - "current_tier_rank": cur_rank, - "upgrade_required": bool(required) and req_rank > cur_rank, - "allowed": allowed, - } - ) - except Exception as exc: - logger.warning("api_entitlement_required_tier: error: %s", exc) - feature = (request.args.get("feature") or "").strip().lower() - runtime = (request.args.get("runtime") or "").strip().lower() - channels_raw = (request.args.get("channels") or "").strip() - retention_raw = (request.args.get("retention_days") or "").strip() - nodes_raw = (request.args.get("nodes") or "").strip() - if feature: - key, kind = feature, "feature" - elif runtime: - key, kind = runtime, "runtime" - elif channels_raw: - key, kind = channels_raw, "channels" - elif retention_raw: - key, kind = retention_raw, "retention_days" - elif nodes_raw: - key, kind = nodes_raw, "nodes" - else: - key, kind = "", "" - return jsonify( - { - "key": key, - "kind": kind, - "required_tier": None, - "required_tier_label": None, - "required_tier_rank": -1, - "current_tier": "oss", - "current_tier_rank": 0, - "upgrade_required": False, - "allowed": True, - } - ) - - -def _has_axis_fallback(axis: str, key: str) -> dict: - """OSS-free / never-5xx shape for the ``/api/entitlement/has-*`` - endpoints, matching the never-crash posture of - ``/api/entitlement/required-tier`` and ``/api/entitlement/lock-reason``. - - Same 8-key envelope as the happy-path branch so a frontend can bind - ``allowed`` off the URL without a branch on the underlying resolver - state. ``axis`` is ``"feature"`` or ``"runtime"`` -- the key name of - the input arg -- so a single helper serves both sibling endpoints. - """ - return { - axis: key, - f"has_{axis}": False, - "allowed": False, - "required_tier": None, - "required_tier_label": None, - "required_tier_rank": -1, - "current_tier": "oss", - "current_tier_rank": 0, - "upgrade_required": False, - } - - -def _has_axis_body(axis: str, resolver_min_tier, resolver_allow) -> dict: - """Happy-path body builder for the ``/api/entitlement/has-*`` - endpoints -- scalar boolean plus the surrounding required-tier - envelope so a paywall tile can bind ``has_feature`` / - ``has_runtime`` directly off the URL without a follow-up hit to - ``/api/entitlement/required-tier``. - - Envelope keys are byte-stable across ``has_feature`` / - ``has_runtime`` (parameterised via ``axis``) and match the tier - columns on the sibling ``/required-tier`` body so a cross-endpoint - consistency invariant (same tier answer for the same key) can be - pinned in tests. - """ - from clawmetry import entitlements as _ent - - key = (request.args.get(axis) or "").strip().lower() - ent = _ent.get_entitlement() - if axis == "feature": - has_flag = _ent.has_feature(key) - required = _ent.min_tier_for_feature(key) if key else None - else: - has_flag = _ent.has_runtime(key) - required = _ent.min_tier_for_runtime(key) if key else None - # `resolver_*` params kept in the signature so tests can monkeypatch - # a single seam if the resolver ever grows a second entry point. - _ = (resolver_min_tier, resolver_allow) - cur_rank = _ent.tier_rank(ent.tier) - req_rank = _ent.tier_rank(required) if required else -1 - required_label = _ent.tier_label(required) if required else None - return { - axis: key, - f"has_{axis}": bool(has_flag), - "allowed": bool(has_flag), - "required_tier": required, - "required_tier_label": required_label, - "required_tier_rank": req_rank, - "current_tier": ent.tier, - "current_tier_rank": cur_rank, - "upgrade_required": bool(required) and req_rank > cur_rank, - } - - -@bp_entitlement.route("/api/entitlement/has-feature") -def api_entitlement_has_feature(): - """``GET /api/entitlement/has-feature?feature=`` -- boolean-gate - scalar sibling of ``/api/entitlement/required-tier?feature=``. - - Returns ONE boolean (``has_feature``) plus the surrounding tier - envelope (``current_tier``, ``required_tier``, ``upgrade_required``) - so a paywall tile can bind ``allowed`` directly off this URL without - parsing the full required-tier body. Grace-safe: while - :attr:`Entitlement.grace` is ``True`` (the current rollout state) - ``has_feature`` reports ``True`` for every KNOWN feature id, so - wiring this into a gate today changes NO current behavior. - Unknown / empty / non-string ids collapse to ``has_feature=False`` - without an HTTP 4xx (the never-crash posture matches the sibling - ``/api/entitlement/required-tier`` and ``/api/entitlement/lock-reason`` - endpoints). Never 5xx. - """ - try: - from clawmetry import entitlements as _ent - - return jsonify( - _has_axis_body( - "feature", - _ent.min_tier_for_feature, - _ent.has_feature, - ) - ) - except Exception as exc: - logger.warning("api_entitlement_has_feature: error: %s", exc) - key = (request.args.get("feature") or "").strip().lower() - return jsonify(_has_axis_fallback("feature", key)) - - -@bp_entitlement.route("/api/entitlement/has-runtime") -def api_entitlement_has_runtime(): - """``GET /api/entitlement/has-runtime?runtime=`` -- runtime-axis - mirror of ``/api/entitlement/has-feature``. - - Same 8-key envelope with ``runtime`` / ``has_runtime`` in the - axis-specific slots. Grace-safe: ``has_runtime`` reports ``True`` - for every known runtime id while grace is on; unknown / empty ids - collapse to ``False``. Never 5xx. - """ - try: - from clawmetry import entitlements as _ent - - return jsonify( - _has_axis_body( - "runtime", - _ent.min_tier_for_runtime, - _ent.has_runtime, - ) - ) - except Exception as exc: - logger.warning("api_entitlement_has_runtime: error: %s", exc) - key = (request.args.get("runtime") or "").strip().lower() - return jsonify(_has_axis_fallback("runtime", key)) - - -def _has_axis_at_fallback(axis: str, tier: str, key: str) -> dict: - """OSS-free / never-5xx shape for ``/api/entitlement/has-feature-at`` / - ``/has-runtime-at``. - - What-if sibling of :func:`_has_axis_fallback`. On any resolver / helper - blowup the endpoint still returns 200 with the same 12-key envelope as - the happy-path branch, but with ``has__at`` and ``allowed`` - strict-``False`` (matches the fail-closed posture the sibling - ``/has-feature`` / ``/has-runtime`` fallback uses -- a paywall matrix - tile that lost the resolver never silently renders a grant it can't - verify). ``tier`` and the axis slot echo the caller's canonicalised - input so the UI can still surface both in a diagnostics tooltip. - """ - return { - "tier": tier, - axis: key, - f"has_{axis}_at": False, - "allowed": False, - "required_tier": None, - "required_tier_label": None, - "required_tier_rank": -1, - "perspective_tier_rank": -1, - "current_tier": "oss", - "current_tier_rank": 0, - "grace": True, - "enforced": False, - } - - -def _has_axis_at_body(axis: str) -> dict: - """Happy-path body builder for ``/api/entitlement/has-feature-at`` / - ``/has-runtime-at`` -- scalar what-if boolean plus the surrounding - what-if envelope. - - Perspective-shaped sibling of :func:`_has_axis_body`: where the live - variant folds :func:`has_feature` / :func:`has_runtime` against the - resolved entitlement, this folds :func:`has_feature_at` / - :func:`has_runtime_at` against a caller-supplied ``tier=`` perspective - so a pricing matrix ("does Starter grant fleet? Pro? Enterprise?") - can bind ONE boolean per cell off ONE URL each instead of - hydrating the full ``/feature-catalog-at`` payload and pulling out - the ``allowed`` field client-side. - - 12-key envelope (adds ``tier`` + ``perspective_tier_rank`` + - ``grace`` / ``enforced`` on top of the sibling ``/has-feature`` shape):: - - { - "tier": "" | "", - "feature": "" | "", - "has_feature_at": , - "allowed": , # alias of has_feature_at - "required_tier": "" | null, # min_tier_for_ - "required_tier_label": "