feat(credits): usage-aware credits — in-session notices, /usage view, dev readout (#40011)
* feat(tui): HERMES_DEV_CREDITS live-spend dev readout (L0 tracer for usage-aware credits)
L0 of the usage-aware-credits feature: a dev-only, env-gated tracer that
exercises the real header -> CreditsState -> TUI pipe end-to-end behind
HERMES_DEV_CREDITS, de-risking the L1/L5 build before the notice policy exists.
- agent/credits_tracker.py: CreditsState + parse_credits_headers (headers are
strings -> paid_access via == "true", never bool(); retain-last-known; only
subscription_micros may be negative; *_usd kept verbatim).
- run_agent.py: _capture_credits / get_credits_state / get_credits_spent_micros,
session-start baseline latch, + dev-gated "credits" capture log.
- agent/chat_completion_helpers.py: capture on the streaming response.
- agent/agent_init.py: init _credits_state + _credits_session_start_micros.
- tui_gateway/server.py: _get_usage emits dev_credits_spent_micros only when flagged.
- ui-tui appChrome.tsx / types.ts: cents delta status segment + "(dev credits)" banner.
Off by default; silent for normal users. Validated live against staging
(capture log delta matches the TUI segment). Throwaway consumer (readout/log/
banner); credits_tracker + the capture plumbing are the real feature foundation.
* test(credits): lock parser under 9-state matrix + harden validation (L2)
Add tests/agent/test_credits_tracker.py with 92 tests covering the 9-state
matrix (healthy, sub_90pct, grant_exhausted, purchased_only, tool_pool_free,
depleted, debt, missing, no_org) plus validation edge cases: version strict==1
with warn-once latch for v>1, bool-string trap (paid_access/tool_pool_gated_off
== "true"/"false", never bool()), half-pair subscription limit treated as
both-absent while parse succeeds, USD regex ^-?\d+\.\d{2}$, non-int micros
→ None, negative non-subscription micros → None, as_of_ms junk → None, zero
limit ZeroDivision guard.
Harden agent/credits_tracker.py to match the spec:
- Add tool_pool_micros/tool_pool_gated_off/from_header fields to CreditsState
- Add depleted property (== not paid_access, never remaining==0)
- Change used_fraction guard to key off subscription_limit_micros (the actual
denominator) not denominator_kind (metadata)
- Replace fail-soft _safe_int with a sentinel-returning variant; full validation
now returns None on any malformed field rather than silently defaulting
- Add module-level warn-once latch for version > 1
- Add USD regex validation; add denominator_kind allow-list check
- Parse x-nous-tool-pool-* prefix headers (not x-nous-credits-tool-pool-*)
* feat(credits): notice spine — AgentNotice + notice_callback/notice_clear_callback + TUI binding (L1)
L1 of usage-aware credits: the driver-agnostic notice delivery spine that L4's
policy will fire through and L5's TUI render will consume.
- agent/credits_tracker.py: AgentNotice dataclass (text/level/kind/ttl_ms/key/id;
kind defaults "sticky", kept TTL-expressive for a future config seam).
- run_agent.py: AIAgent gains notice_callback + notice_clear_callback slots and
_emit_notice / _emit_notice_clear emitters (swallow all callback errors — a
notice must never break the agent loop; no-op when unbound).
- agent/agent_init.py: thread both callbacks through init_agent.
- tui_gateway/server.py: bind both in _agent_cbs → notification.show / notification.clear
WS events (snake_case payload, matching the existing gateway-event convention).
- ui-tui/src/gatewayTypes.ts: notification.show / notification.clear arms on GatewayEvent.
- tests/run_agent/test_notice_spine.py: 15 tests (emitter fire + fail-open + no-op,
signature threading, TUI binding payload shape).
Messaging push is out of v1 (binds neither callback). CLI binding + the TUI render/
decode land with L4 (firing) and L5 (render) so turn-end flush is wired correctly.
* feat(credits): threshold reconciliation policy + tests (L4.1)
* feat(credits): wire threshold policy into capture + latch (L4.2)
After a fresh header parse, _capture_credits runs evaluate_credits_notices against
the agent's _credits_latch and emits the result — clears first, then shows (so a
recovered depletion clears before the "restored" success lands, and depleted wins
the latest-wins slot). Gated on a bound notice_callback: messaging (no callbacks)
still caches state for /usage but runs no policy. Parse stays fail-open (miss →
keep last-known); the eval/emit path warns on failure rather than swallowing, so a
depletion-notice bug can't vanish silently.
- run_agent.py: _capture_credits split into parse (swallow→miss) + policy (warn);
latch lazy-guarded (object.__new__ safety).
- agent/agent_init.py: init agent._credits_latch = {"active": set(), "seen_below_90": False}.
* feat(tui): render credits notices in the status bar (L5, Strategy B)
The TUI now renders the notification.show / notification.clear gateway events the
agent emits — a level-colored notice overrides the status/verb slot when not busy.
- Notice state machine on turnController (pendingNotice + dedicated noticeTimer +
show/clear/applyNotice/flushPendingNotice/clearNoticeState). createGatewayEventHandler
decodes the events and delegates.
- Render priority busy > notice > status (appChrome StatusRule); notice text rendered
verbatim (its glyph comes from the policy), shrinkable so it never clips model│ctx;
dev-credits banner + Δ segment preserved. UiState.notice is snake_case (matches wire).
- Busy-wins: a notice arriving mid-turn is held and flushed at the THREE turn-end sites
(recordMessageComplete / interruptTurn / recordError) — never idle(), which reset()
also calls (would leak across sessions); reset() clears instead.
- Dedicated noticeTimer (never statusTimer); TTL starts on visibility with an id-guard;
latest-wins cancels the prior timer; clear is key-matched (no-op on mismatch); a sticky
survives a turn (flush no-ops with no pending); session reset clears (no cross-session leak).
- 20 tests (handler/turnController logic incl. R3-C2 timer isolation + render priority).
* feat(credits): cold-start seed for new Nous sessions (L3)
A genuinely-new Nous session has no inference header yet, so seed credits state from
the authoritative GET /api/oauth/account snapshot at session start (in the new-session
branch of _restore_or_build_system_prompt — inline, since the on_session_start plugin
hook gets no agent reference). The seed runs the shared notice policy, so a session that
opens already depleted warns IMMEDIATELY rather than only after the first turn.
- Maps the nested account fields (paid_service_access → paid_access; total_usable /
subscription / purchased on paid_service_access_info; rollover on subscription), each
None-guarded; float dollars → micros via round(d*1e6), *_usd left "" (render formats
from micros — never synthesize a verbatim usd from a float).
- Magnitudes-only: no monthlyCredits on the endpoint → subscription_limit_* unset →
used_fraction None → no warn90 from the seed (% only once a header lands, per D-E).
- Provider-guarded to Nous; fail-open (any error leaves _credits_state None, never
blocks startup); paid_access unknown ⇒ True (never falsely depleted).
- run_agent.py: extracted the warm-path policy/emit block into a shared
_emit_credits_notices() so capture and the seed fire notices identically.
* feat(credits): /usage Nous credits magnitudes view + recovery trigger (L6)
Add Nous credit dollar magnitudes to /usage (subscription / top-up / total
+ rollover + renewal + portal CTA), magnitudes-only per v1 (no % until the
account endpoint exposes a denominator). Reuses the existing account-usage
render machinery via a new pure build_nous_credits_snapshot() that maps a
NousPortalAccountInfo to an AccountUsageSnapshot; no nous branch is added to
fetch_account_usage (keeps the per-provider boundary intact).
CLI /usage also doubles as a depletion-recovery trigger: a force_fresh
account fetch, kept in a SEPARATE local so it never clobbers the
header-sourced agent._credits_state (which alone carries used_fraction). If
paid access recovered while credits.depleted is latched and a notice
consumer is bound, it reuses agent._emit_credits_notices() to clear it.
Gateway /usage displays magnitudes only — messaging binds no notice
consumer, so it performs no recovery emit.
Fail-open throughout: any portal hiccup leaves /usage unaffected.
* refactor(credits): dedupe HERMES_DEV_CREDITS flag parse via shared helpers
The dev-flag truthy check was inlined in three places. Replace with the shared
utils.is_truthy_value (run_agent.py, tui_gateway/server.py — also drops a
redundant inline `import os`) and a hoisted DEV_CREDITS_MODE export in
ui-tui/src/config/env.ts (consumed by appChrome, which also stops recomputing the
env check on every render). Behaviour-preserving; identical truthy set.
* fix(credits): cut dead /usage recovery trigger + bound portal fetches (L6 review)
Adversarial review found the /usage depletion-recovery trigger dead AND broken:
the CLI binds no notice_clear_callback, the TUI runs /usage in a separate
slash-worker subprocess (its own agent/latch), and the no-clobber rule made it
evaluate stale paid_access anyway. Recovery already happens on the next inference
(warm path), so the trigger was redundant — remove it and stop the depleted
notice over-promising.
- cli.py: remove the dead recovery block; bound the /usage portal fetch with a
10s wall-clock timeout (ThreadPoolExecutor) like the per-provider fetch —
urllib's per-socket timeout is not a wall-clock guarantee.
- agent/credits_tracker.py: reword the depleted CTA to "run /usage for balance"
(no false recovery promise; /usage shows fresh magnitudes, sticky clears next turn).
- agent/conversation_loop.py: same wall-clock timeout on the cold-start seed fetch
so a stalled portal can't hang session startup; tidy its time import.
* chore(credits): dev notice-state fixtures (HERMES_DEV_CREDITS_FIXTURE)
Throwaway dev scaffolding to exercise the notice pipeline without real spend or
Redis seeding. Set HERMES_DEV_CREDITS_FIXTURE to a state name (healthy / sub_90pct
/ grant_exhausted / depleted / clear) or a file path whose contents name a state
(re-read each turn → flip states live for recovery testing). _capture_credits
injects the chosen CreditsState instead of parsing real headers and runs the
shared notice policy. Deletable with the rest of the HERMES_DEV_CREDITS scaffolding.
* feat(credits): /usage monthly-grant % gauge
The portal /api/oauth/account subscription block now carries monthly_credits
(the per-period grant allowance, the % denominator). The consumer parsed
monthly_charge but dropped monthly_credits, so /usage stayed magnitudes-only.
Capture monthly_credits into NousPortalSubscriptionInfo + _subscription_from_payload.
build_nous_credits_snapshot emits a Subscription usage window (real % used, routed
through the existing render machinery) when monthly_credits is a finite positive
denominator and credits_remaining is finite and <= cap; otherwise it degrades to
magnitudes-only (older portals, rollover-over-cap, or non-finite payloads).
Guards (adversarial-review-driven): reject non-finite operands (json.loads parses
bare NaN/Infinity by default → would render $nan + a false 100% used), reject
bools, guard div-by-zero (cap>0), and suppress the gauge when remaining > cap
(rollover spanning the period makes the cap a nonsensical denominator → the
$X-of-$Y detail would read as a contradiction). Debt (remaining<0) clamps to 100%.
Money rule preserved: the ratio + magnitudes are computed from numeric float
account fields via display formatting, never by parsing a server *_usd string
(there are none on these dataclasses).
13 gauge tests added (tests/agent/test_nous_credits_gauge.py).
* fix(credits): show /usage Nous block whenever a Nous account is present
/usage runs in a slash-worker subprocess whose resolved inference provider is
often not "nous" even when the user has a Nous account, so gating the Nous
credits block on (provider == "nous") hid it entirely — the account data was
fully available but never rendered.
Gate instead on "a Nous account is logged in": a cheap local auth-state lookup
(get_provider_auth_state('nous') has an access_token) decides whether to attempt
the portal fetch, regardless of which provider inference runs on. In the gateway
the block is also lifted out of the 'if provider:' scope so a Nous-credentialled
user with another (or no) resident inference provider still sees their balance.
Fail-open and the per-fetch wall-clock timeout are preserved.
* fix(credits): show /usage Nous block when there's no live agent (TUI slash-worker)
In the TUI, /usage runs in a slash-worker subprocess that resumes the session
WITHOUT building an agent (self.agent is None), so _show_usage early-returned
"(._.) No active agent" before ever reaching the Nous credits block — which is
agent-independent (a portal fetch gated on Nous auth-state). Extract the block
into _print_nous_credits_block() and run it at the no-agent / no-calls
early-returns too (returns True if it printed, so the fallback message only
shows when there's genuinely nothing).
Verified live against staging: the block + monthly-grant gauge now render in the
slash-worker /usage path (previously hidden). The plain CLI REPL + messaging
paths are unchanged (they have a live agent).
* feat(credits): escalating 50/75/90 usage bands (single status line)
Replace the lone 90%-used warning with three escalating bands (50 info, 75 warn,
90 warn) shown as ONE status-bar line: it displays the highest band the
subscription grant has crossed, replaces the line as usage climbs, steps back
down on recovery, and clears below 50%. No stacking, no per-turn churn.
Bands live in a tunable CREDITS_USAGE_BANDS list; the policy derives everything
from it. Single notice key (credits.usage) with a usage_band latch field so the
notice only re-emits when the band actually changes. The crossing gate
(seen_below_90) is preserved so a fresh live session that opens mid-range stays
quiet until it has been observed below the lowest band (cold-start primes it when
it wants an open-high warning). Denominator math unchanged: % = subscription
grant burn (cap - grant_remaining)/cap, clamped [0,1]; top-up never moves the %.
Migrated test_credits_policy.py to the new key + added TestUsageBands (climb,
step-down, recovery-clear, idempotent, inclusive boundaries).
* feat(credits): hydrate notices at session OPEN via shared seed (TUI + first-turn)
Notices previously only fired inside a conversation turn (first message), so a
session that opened already depleted / past a usage band showed nothing at
'ready'. Extract the cold-start seed into a shared seed_credits_at_session_start()
and call it (a) in the TUI/desktop agent build right after the notice callback is
wired (fires at 'ready', before any message) and (b) as the first-turn fallback in
conversation_loop. Idempotent (skips once _credits_state exists) and fail-open.
The seed now maps monthly_credits -> subscription_limit_micros +
denominator_kind='subscription_cap', so used_fraction is computable at seed time
and usage-band warnings (not just depletion) hydrate on open. Primes the crossing
latch so a session opening already in a band warns immediately. Degrades to
depletion-only when monthly_credits is absent (older portals).
Adds test_credits_cold_start.py covering open-at-band, depletion, debt, no-cap
degradation, and the shared seed (fires/idempotent/skips-non-nous).
* feat(credits): /usage monthly-grant % gauge + fixture support + TUI surfacing
agent/account_usage.py: build_nous_credits_snapshot emits a subscription %% gauge
when the portal supplies a positive, finite monthly_credits denominator with
remaining <= cap (guards reject NaN/Infinity and rollover-over-cap, which would
render $nan or a contradictory $X-of-$Y); degrades to magnitudes-only otherwise.
Adds shared nous_credits_lines() (auth-gated, wall-clock-bounded portal fetch) so
the CLI and TUI /usage render the same block, and _snapshot_from_credits_state()
so HERMES_DEV_CREDITS_FIXTURE drives /usage offline too.
TUI: session.usage RPC carries credits_lines (agent-independent) and the /usage
panel renders them regardless of API-call count or resume state — previously the
TUI's separate /usage implementation only showed token counts.
Money rule preserved: %% and magnitudes come from numeric float account fields via
display formatting, never by parsing a server *_usd string.
* feat(credits): CLI REPL inline notices (parity with TUI)
The plain CLI agent bound no notice callbacks, so credit notices were TUI-only.
Bind notice_callback/notice_clear_callback on the CLI AIAgent; _on_notice renders
a single level-colored line above the prompt (error red / warn yellow / success
green / info dim) via _cprint, and seed credits at session open so a depletion or
usage-band warning shows before the first message — the same hydration the TUI
got. _on_notice_clear is a no-op (the REPL prints lines, no persistent slot).
* test(credits): add sub_50pct + sub_75pct dev fixtures for the new usage bands
The fixture set jumped 10%% -> 90%%; add sub_50pct (uf 0.5 -> band 50 info) and
sub_75pct (uf 0.75 -> band 75 warn) so the new escalating bands are exercisable
via HERMES_DEV_CREDITS_FIXTURE across all three surfaces (notice, session-open
seed, /usage gauge).
* fix(credits): usage-band notice clears on next prompt (not sticky-forever)
A 50/75/90 usage heads-up was sticky and camped the status bar indefinitely. Clear
the visible credits.usage notice when a new turn starts (startMessage), so it shows
until your next prompt then yields. The server latch is unchanged, so it won't
re-nag at the same band — it only re-shows when the band actually changes (climb)
or clears when usage drops below the lowest band. Depletion stays sticky.
* refactor(credits): consolidate the /usage credits block behind nous_credits_lines()
The CLI (_print_nous_credits_block) and the messaging gateway (_handle_usage_command)
each re-implemented the auth-gate + portal fetch + render, and both bypassed the
dev-fixture short-circuit that only the TUI honored — so /usage ignored
HERMES_DEV_CREDITS_FIXTURE on the CLI and in chat. Route both through the shared
agent.account_usage.nous_credits_lines() helper: one fetch/render path, one auth
gate, and the fixture works on every surface (~60 fewer duplicated lines).
The gateway usage test recorded only the last asyncio.to_thread call; /usage now
dispatches both the account fetch and the credits fetch, so it records every call
and matches the account fetch by its provider arg.
* fix(credits): keep the /usage gauge type-safe and log its fail-open path
_is_finite_num is now a TypeGuard[float], so the type checker narrows the gauge
operands (monthly_credits / credits_remaining) and the magnitudes passed to
_fmt_usd through it — no more None-operand warnings on the arithmetic. Add a debug
breadcrumb on the nous_credits_lines portal-fetch fail-open so a dead /usage block
is diagnosable in agent.log without a dev flag.
* fix(credits): harden the header tracker — prod-leak gate, hot-path probe, fire-and-forget seed
- Prod-leak guard: dev fixtures (HERMES_DEV_CREDITS_FIXTURE) now also require
HERMES_DEV_CREDITS, so a stray fixture var can't surface fabricated balances on a
real account. Matches the documented run workflow (both vars set together).
- Hot-path probe: parse_credits_headers checks for the version sentinel header
before allocating a lowercased copy of the response headers — skips that work on
every non-Nous API call. Behaviour-identical and still case-insensitive.
- Fire-and-forget seed: the real portal fetch in seed_credits_at_session_start now
runs in a daemon thread, so a slow/unreachable portal never delays session "ready"
(previously blocked up to 10s). The dev-fixture path stays synchronous; the thread
re-checks idempotency before hydrating (a live header may land first).
- Diagnostics: debug breadcrumbs on the parse and seed fail-open paths so a crashed
parser / dead seed is distinguishable from a legitimate no-headers miss.
Cold-start tests set HERMES_DEV_CREDITS alongside the fixture to match the gate.
* test(tui): fix env-timing in the StatusRule dev-credits assertion
DEV_CREDITS_MODE is read once at module load (config/env), so mutating
process.env.HERMES_DEV_CREDITS inside the test couldn't flip it — the dev-banner
assertion only passed if the env was exported before vitest started, and failed in a
normal run. Move that assertion to a sibling file that mocks config/env with
DEV_CREDITS_MODE: true (scoped, no module-reset / React-identity hazard).
* test(credits): cover the dev-fixture /usage render and usage-band clear-on-prompt
- _snapshot_from_credits_state (the offline /usage renderer) had no direct test:
lock the gauge math, the verbatim *_usd magnitudes, the depletion line and the
fixture marker, plus the no-cap (no gauge) and None-state cases.
- turnController.startMessage had no test for clearing the credits.usage notice on
the next prompt while leaving credits.depleted sticky.
* feat(credits): deliver credit notices over messaging gateways
Bind notice_callback/notice_clear_callback on the per-turn gateway agent
so usage-band / depletion / restored notices reach Telegram/Discord/Slack/
etc. Previously the messaging gateway bound neither callback, so the agent's
_emit_credits_notices early-returned and a chat user crossing a band got
nothing unless they ran /usage manually.
- render_notice_line(): AgentNotice -> single plaintext line (level glyph +
text), plaintext-only so it renders uniformly without per-platform escaping.
Fail-soft on malformed/empty notices.
- Standalone push for every notice (messaging has no persistent status bar):
route through the shared _deliver_platform_notice rail (honors private/
public delivery + thread metadata), scheduled onto the gateway loop via
safe_schedule_threadsafe from the agent's sync worker thread — same pattern
as _status_callback_sync.
- The fired-once latch lives on the cached (reused-in-place) agent and
persists across turns, so a band crosses once -> one push, no per-turn
re-nag. Re-fires only after idle-eviction rebuilds the agent (a reminder).
- Recovery ('Credit access restored') rides the show path (emitted as a
success notice, not a clear). notice_clear_callback is a no-op: a sent
platform message can't be cleanly retracted.
Tests: render glyph/levels/fail-soft + public/private delivery seam through
_deliver_platform_notice + no-adapter no-op.
* fix(credits): don't double the glyph on messaging notices
render_notice_line prepended a per-level glyph, but the notice policy already
bakes the glyph into the text (and the TUI + CLI render it verbatim) — so every
credit notice over messaging came out doubled ("⚠ ⚠ Credits 90% used",
"⛔ ✕ Credit access paused"). Emit the text verbatim instead; drop the now-dead
level→glyph map.
The render tests fed glyph-less text (and the success case only checked
startswith), so the doubling slipped through. Rework them around the verbatim
contract and add an end-to-end regression that runs real evaluate_credits_notices
output through render_notice_line and asserts the line is returned unchanged.
This commit is contained in:
Siddharth Balyan
@@ -0,0 +1,192 @@
|
||||
"""Tests for cold-start credits hydration at session open.
|
||||
|
||||
The L3 cold-start seed primes agent._credits_state from /api/oauth/account (or a
|
||||
HERMES_DEV_CREDITS_FIXTURE) so depletion AND the 90% grant warning fire immediately
|
||||
at session open, not only after the first inference header. These tests assert the
|
||||
notice policy fires correctly for a seed-shaped CreditsState with the warn90 latch
|
||||
primed the way conversation_loop does it.
|
||||
"""
|
||||
import time
|
||||
|
||||
from agent.credits_tracker import CreditsState, evaluate_credits_notices
|
||||
|
||||
|
||||
def _cold_start_notices(state: CreditsState):
|
||||
"""Mirror the conversation_loop seed: prime seen_below_90 when used_fraction is
|
||||
computable (the snapshot IS the first observation), then evaluate once."""
|
||||
latch = {"active": set(), "seen_below_90": False}
|
||||
if state.used_fraction is not None:
|
||||
latch["seen_below_90"] = True
|
||||
show, clear = evaluate_credits_notices(state, latch)
|
||||
return [n.key for n in show]
|
||||
|
||||
|
||||
def _state(**kw) -> CreditsState:
|
||||
kw.setdefault("from_header", False)
|
||||
kw.setdefault("captured_at", time.time())
|
||||
return CreditsState(**kw)
|
||||
|
||||
|
||||
def test_cold_start_healthy_no_notice():
|
||||
s = _state(
|
||||
remaining_micros=30_340_000, subscription_micros=18_000_000,
|
||||
subscription_limit_micros=20_000_000, subscription_limit_usd="20.00",
|
||||
denominator_kind="subscription_cap", paid_access=True,
|
||||
)
|
||||
assert abs(s.used_fraction - 0.1) < 1e-9
|
||||
assert _cold_start_notices(s) == []
|
||||
|
||||
|
||||
def test_cold_start_opens_already_at_90pct_warns():
|
||||
"""A session that OPENS already ≥90% must warn immediately — the seed primes
|
||||
seen_below_90 so warn90 fires without a prior live crossing."""
|
||||
s = _state(
|
||||
remaining_micros=2_000_000, subscription_micros=2_000_000,
|
||||
subscription_limit_micros=20_000_000, subscription_limit_usd="20.00",
|
||||
denominator_kind="subscription_cap", paid_access=True,
|
||||
)
|
||||
assert s.used_fraction == 0.9
|
||||
assert "credits.usage" in _cold_start_notices(s)
|
||||
|
||||
|
||||
def test_cold_start_grant_exhausted_warns_and_grant_spent():
|
||||
s = _state(
|
||||
remaining_micros=12_340_000, subscription_micros=0,
|
||||
subscription_limit_micros=20_000_000, subscription_limit_usd="20.00",
|
||||
purchased_micros=12_340_000, denominator_kind="subscription_cap", paid_access=True,
|
||||
)
|
||||
assert s.used_fraction == 1.0
|
||||
keys = _cold_start_notices(s)
|
||||
assert "credits.usage" in keys
|
||||
assert "credits.grant_spent" in keys
|
||||
|
||||
|
||||
def test_cold_start_depleted_warns():
|
||||
s = _state(
|
||||
remaining_micros=0, subscription_micros=0, purchased_micros=0,
|
||||
paid_access=False, disabled_reason="out_of_credits",
|
||||
)
|
||||
assert s.used_fraction is None # no cap → no %, depletion keys off paid_access
|
||||
assert _cold_start_notices(s) == ["credits.depleted"]
|
||||
|
||||
|
||||
def test_cold_start_debt_warns_and_depleted():
|
||||
"""Negative subscription balance (the only signed field) → 100% used + depleted."""
|
||||
s = _state(
|
||||
remaining_micros=0, subscription_micros=-5_000_000,
|
||||
subscription_limit_micros=20_000_000, subscription_limit_usd="20.00",
|
||||
denominator_kind="subscription_cap", paid_access=False,
|
||||
disabled_reason="out_of_credits",
|
||||
)
|
||||
assert s.used_fraction == 1.0
|
||||
keys = _cold_start_notices(s)
|
||||
assert "credits.usage" in keys
|
||||
assert "credits.depleted" in keys
|
||||
|
||||
|
||||
def test_cold_start_no_cap_degrades_to_depletion_only():
|
||||
"""Without monthly_credits (older portals) the seed sets no limit → used_fraction
|
||||
None → only depletion can fire, never warn90."""
|
||||
healthy_no_cap = _state(
|
||||
remaining_micros=30_000_000, subscription_micros=18_000_000,
|
||||
subscription_limit_micros=None, denominator_kind="none", paid_access=True,
|
||||
)
|
||||
assert healthy_no_cap.used_fraction is None
|
||||
assert _cold_start_notices(healthy_no_cap) == []
|
||||
|
||||
|
||||
def test_dev_fixtures_drive_cold_start():
|
||||
"""Every HERMES_DEV_CREDITS_FIXTURE state produces a valid seed CreditsState."""
|
||||
import os
|
||||
|
||||
from agent.credits_tracker import dev_fixture_credits_state
|
||||
|
||||
expected = {
|
||||
"healthy": [],
|
||||
"sub_90pct": ["credits.usage"],
|
||||
"depleted": ["credits.depleted"],
|
||||
}
|
||||
for name, want in expected.items():
|
||||
os.environ["HERMES_DEV_CREDITS"] = "1" # fixtures gate on the dev flag
|
||||
os.environ["HERMES_DEV_CREDITS_FIXTURE"] = name
|
||||
try:
|
||||
fx = dev_fixture_credits_state()
|
||||
assert fx is not None, name
|
||||
assert _cold_start_notices(fx) == want, (name, _cold_start_notices(fx))
|
||||
finally:
|
||||
os.environ.pop("HERMES_DEV_CREDITS_FIXTURE", None)
|
||||
os.environ.pop("HERMES_DEV_CREDITS", None)
|
||||
|
||||
|
||||
# ── seed_credits_at_session_start: the shared session-open hydrator ───────────
|
||||
|
||||
|
||||
class _FakeAgent:
|
||||
"""Minimal agent surface for the seed helper: state slots + an emit that runs
|
||||
the real policy against the latch."""
|
||||
|
||||
def __init__(self, provider="nous"):
|
||||
from agent.credits_tracker import evaluate_credits_notices
|
||||
|
||||
self.provider = provider
|
||||
self._credits_state = None
|
||||
self._credits_session_start_micros = None
|
||||
self._credits_latch = {"active": set(), "seen_below_90": False, "usage_band": None}
|
||||
self.emitted: list = []
|
||||
self._eval = evaluate_credits_notices
|
||||
|
||||
def _emit_credits_notices(self):
|
||||
if self._credits_state is None:
|
||||
return
|
||||
show, clear = self._eval(self._credits_state, self._credits_latch)
|
||||
self.emitted.append(([n.key for n in show], clear))
|
||||
|
||||
|
||||
def _seed(agent, fixture):
|
||||
import os
|
||||
|
||||
from agent.credits_tracker import seed_credits_at_session_start
|
||||
|
||||
os.environ["HERMES_DEV_CREDITS"] = "1" # fixtures gate on the dev flag
|
||||
os.environ["HERMES_DEV_CREDITS_FIXTURE"] = fixture
|
||||
try:
|
||||
return seed_credits_at_session_start(agent)
|
||||
finally:
|
||||
os.environ.pop("HERMES_DEV_CREDITS_FIXTURE", None)
|
||||
os.environ.pop("HERMES_DEV_CREDITS", None)
|
||||
|
||||
|
||||
def test_seed_fires_usage_band_at_session_open():
|
||||
a = _FakeAgent()
|
||||
assert _seed(a, "sub_90pct") is True
|
||||
assert a._credits_state is not None
|
||||
assert a.emitted == [(["credits.usage"], [])]
|
||||
|
||||
|
||||
def test_seed_fires_depleted_at_session_open():
|
||||
a = _FakeAgent()
|
||||
assert _seed(a, "depleted") is True
|
||||
assert a.emitted == [(["credits.depleted"], [])]
|
||||
|
||||
|
||||
def test_seed_healthy_no_notice():
|
||||
a = _FakeAgent()
|
||||
assert _seed(a, "healthy") is True
|
||||
assert a.emitted == [([], [])]
|
||||
|
||||
|
||||
def test_seed_is_idempotent():
|
||||
a = _FakeAgent()
|
||||
_seed(a, "sub_90pct")
|
||||
a.emitted = []
|
||||
# second call must no-op (state already populated)
|
||||
assert _seed(a, "sub_90pct") is False
|
||||
assert a.emitted == []
|
||||
|
||||
|
||||
def test_seed_skips_non_nous():
|
||||
from agent.credits_tracker import seed_credits_at_session_start
|
||||
|
||||
a = _FakeAgent(provider="openrouter")
|
||||
assert seed_credits_at_session_start(a) is False
|
||||
assert a._credits_state is None
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Tests for _snapshot_from_credits_state — the dev-fixture /usage renderer.
|
||||
|
||||
``build_nous_credits_snapshot`` maps a live portal account; ``_snapshot_from_credits_state``
|
||||
maps a header-shaped CreditsState (e.g. a HERMES_DEV_CREDITS_FIXTURE) into the SAME
|
||||
/usage snapshot shape, so the gauge + magnitudes are exercisable offline. These lock
|
||||
the gauge math, the verbatim *_usd magnitudes (never parseFloat'd), the depletion line,
|
||||
and the dev-fixture marker.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from agent.account_usage import _snapshot_from_credits_state
|
||||
from agent.credits_tracker import CreditsState
|
||||
|
||||
|
||||
def _state(**kw) -> CreditsState:
|
||||
kw.setdefault("from_header", True)
|
||||
return CreditsState(**kw)
|
||||
|
||||
|
||||
def test_renders_gauge_magnitudes_and_fixture_marker():
|
||||
# used_fraction = (20 - 10) / 20 = 0.5 → a 50%-used gauge window
|
||||
snap = _snapshot_from_credits_state(_state(
|
||||
remaining_micros=30_340_000, remaining_usd="30.34",
|
||||
subscription_micros=10_000_000, subscription_usd="10.00",
|
||||
subscription_limit_micros=20_000_000, subscription_limit_usd="20.00",
|
||||
purchased_micros=12_340_000, purchased_usd="12.34",
|
||||
denominator_kind="subscription_cap", paid_access=True,
|
||||
))
|
||||
assert snap is not None and snap.provider == "nous"
|
||||
|
||||
win = next(w for w in snap.windows if w.label == "Subscription")
|
||||
assert win.used_percent is not None and abs(win.used_percent - 50.0) < 1e-9
|
||||
assert win.detail == "$10.00 of $20.00 left" # verbatim *_usd strings, not math
|
||||
|
||||
details = list(snap.details)
|
||||
assert "Subscription credits: $10.00" in details
|
||||
assert "Top-up credits: $12.34" in details
|
||||
assert "Total usable: $30.34" in details
|
||||
assert any("dev fixture" in d for d in details) # the offline marker
|
||||
assert all("access depleted" not in d for d in details)
|
||||
|
||||
|
||||
def test_depleted_adds_status_line():
|
||||
snap = _snapshot_from_credits_state(_state(
|
||||
remaining_micros=0, remaining_usd="0.00",
|
||||
subscription_micros=0, subscription_usd="0.00",
|
||||
purchased_micros=0, purchased_usd="0.00",
|
||||
denominator_kind="none", paid_access=False,
|
||||
))
|
||||
assert snap is not None
|
||||
assert any("access depleted" in d for d in snap.details)
|
||||
|
||||
|
||||
def test_no_cap_yields_no_gauge_window():
|
||||
# No subscription cap → used_fraction is None → no gauge window, magnitudes only.
|
||||
snap = _snapshot_from_credits_state(_state(
|
||||
remaining_micros=5_000_000, remaining_usd="5.00",
|
||||
subscription_micros=5_000_000, subscription_usd="5.00",
|
||||
subscription_limit_micros=None, denominator_kind="none", paid_access=True,
|
||||
))
|
||||
assert snap is not None
|
||||
assert all(w.label != "Subscription" for w in snap.windows)
|
||||
assert "Total usable: $5.00" in snap.details
|
||||
|
||||
|
||||
def test_none_state_is_safe():
|
||||
assert _snapshot_from_credits_state(None) is None
|
||||
@@ -0,0 +1,492 @@
|
||||
"""Tests for evaluate_credits_notices — pure threshold reconciliation policy (L4.1).
|
||||
|
||||
All tests use fresh latch = {"active": set(), "seen_below_90": False, "usage_band": None} per scenario.
|
||||
CreditsState is constructed directly (not parsed from headers).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from agent.credits_tracker import (
|
||||
CREDITS_NOTICE_KIND,
|
||||
CREDITS_RESTORED_TTL_MS,
|
||||
AgentNotice,
|
||||
CreditsState,
|
||||
evaluate_credits_notices,
|
||||
)
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def fresh_latch() -> dict:
|
||||
return {"active": set(), "seen_below_90": False, "usage_band": None}
|
||||
|
||||
|
||||
def state_with_fraction(
|
||||
uf: float | None,
|
||||
*,
|
||||
paid_access: bool = True,
|
||||
denominator_kind: str = "subscription_cap",
|
||||
purchased_micros: int = 0,
|
||||
purchased_usd: str = "0.00",
|
||||
subscription_limit_usd: str | None = "20.00",
|
||||
) -> CreditsState:
|
||||
"""Build a minimal CreditsState that yields the desired used_fraction.
|
||||
|
||||
used_fraction = (limit - subscription_micros) / limit
|
||||
|
||||
When uf is None, we set limit to None so used_fraction returns None.
|
||||
"""
|
||||
if uf is None:
|
||||
return CreditsState(
|
||||
subscription_limit_micros=None,
|
||||
subscription_limit_usd=None,
|
||||
subscription_micros=0,
|
||||
denominator_kind="none",
|
||||
paid_access=paid_access,
|
||||
purchased_micros=purchased_micros,
|
||||
purchased_usd=purchased_usd,
|
||||
)
|
||||
# We want (limit - sub) / limit == uf → sub = limit * (1 - uf)
|
||||
limit = 20_000_000 # $20 in micros
|
||||
sub = int(limit * (1.0 - uf))
|
||||
return CreditsState(
|
||||
subscription_limit_micros=limit,
|
||||
subscription_limit_usd=subscription_limit_usd,
|
||||
subscription_micros=sub,
|
||||
denominator_kind=denominator_kind,
|
||||
paid_access=paid_access,
|
||||
purchased_micros=purchased_micros,
|
||||
purchased_usd=purchased_usd,
|
||||
)
|
||||
|
||||
|
||||
# ── Scenario 1: crossing 90% threshold ───────────────────────────────────────
|
||||
|
||||
|
||||
class TestWarn90Crossing:
|
||||
def test_below_lowest_band_no_notice_but_latch_set(self):
|
||||
latch = fresh_latch()
|
||||
s = state_with_fraction(0.10) # below the 50% band
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
assert all(n.key != "credits.usage" for n in to_show)
|
||||
assert "credits.usage" not in to_clear
|
||||
assert latch["seen_below_90"] is True
|
||||
|
||||
def test_crossing_to_90_fires_once(self):
|
||||
latch = fresh_latch()
|
||||
# First call: uf < 0.5 — sets seen_below_90 (below lowest band)
|
||||
s1 = state_with_fraction(0.10)
|
||||
evaluate_credits_notices(s1, latch)
|
||||
# Second call: uf >= 0.9 — should fire the usage band at 90
|
||||
s2 = state_with_fraction(0.95)
|
||||
to_show, to_clear = evaluate_credits_notices(s2, latch)
|
||||
keys = [n.key for n in to_show]
|
||||
assert "credits.usage" in keys
|
||||
assert "credits.usage" not in to_clear
|
||||
|
||||
def test_no_refire_on_repeated_over_90(self):
|
||||
latch = fresh_latch()
|
||||
s_below = state_with_fraction(0.10)
|
||||
evaluate_credits_notices(s_below, latch)
|
||||
s_over = state_with_fraction(0.95)
|
||||
evaluate_credits_notices(s_over, latch)
|
||||
# Third call: still ≥ 0.9 — must NOT re-fire
|
||||
to_show, to_clear = evaluate_credits_notices(s_over, latch)
|
||||
assert all(n.key != "credits.usage" for n in to_show)
|
||||
assert "credits.usage" not in to_clear
|
||||
|
||||
|
||||
# ── Scenario 2: recovery + re-cross ──────────────────────────────────────────
|
||||
|
||||
|
||||
class TestWarn90RecoveryReCross:
|
||||
def test_recovery_clears_warn90(self):
|
||||
latch = fresh_latch()
|
||||
# Cross below → above
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
evaluate_credits_notices(state_with_fraction(0.95), latch)
|
||||
# Recovery: uf drops back below ALL bands → usage notice clears entirely
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
assert "credits.usage" in to_clear
|
||||
assert "credits.usage" not in latch["active"]
|
||||
|
||||
def test_recross_after_recovery_fires_again(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
evaluate_credits_notices(state_with_fraction(0.95), latch)
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch) # recovery
|
||||
# Re-cross: uf >= 0.9 again — should fire again because the band is clearable
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.95), latch)
|
||||
keys = [n.key for n in to_show]
|
||||
assert "credits.usage" in keys
|
||||
|
||||
|
||||
# ── Scenario 3: open-already-over (hybrid Q3 gate) ───────────────────────────
|
||||
|
||||
|
||||
class TestOpenAlreadyOver:
|
||||
def test_warn90_does_not_fire_without_seen_below_90(self):
|
||||
"""First call uf≥0.9 with seen_below_90=False — warn90 must NOT fire."""
|
||||
latch = fresh_latch()
|
||||
assert latch["seen_below_90"] is False
|
||||
s = state_with_fraction(0.95)
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
assert all(n.key != "credits.usage" for n in to_show)
|
||||
assert "credits.usage" not in to_clear
|
||||
|
||||
|
||||
# ── Scenario 3b: boundary — exact 0.9 and just-below-1.0 ────────────────────
|
||||
|
||||
|
||||
class TestBoundaryFractions:
|
||||
def test_exact_0_9_fires_warn90(self):
|
||||
"""used_fraction == 0.9 exactly must fire warn90 (threshold is inclusive)."""
|
||||
latch = fresh_latch()
|
||||
# First: prime seen_below_90 with a sub-50% observation
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
# Now construct a state where used_fraction is EXACTLY 0.9:
|
||||
# subscription_limit_micros=20_000_000, subscription_micros=2_000_000
|
||||
# → used = 18_000_000 / 20_000_000 = 0.9 exactly
|
||||
s = CreditsState(
|
||||
subscription_limit_micros=20_000_000,
|
||||
subscription_limit_usd="20.00",
|
||||
subscription_micros=2_000_000,
|
||||
denominator_kind="subscription_cap",
|
||||
paid_access=True,
|
||||
)
|
||||
assert s.used_fraction == 0.9
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
keys = [n.key for n in to_show]
|
||||
assert "credits.usage" in keys
|
||||
assert "credits.usage" not in to_clear
|
||||
|
||||
def test_just_below_1_0_does_not_fire_grant_spent(self):
|
||||
"""subscription_micros = limit - 1 (used_fraction just under 1.0) must NOT fire grant_spent.
|
||||
|
||||
Locks the boundary so a future used_fraction clamp refactor cannot fire
|
||||
grant_spent a micro early.
|
||||
"""
|
||||
latch = fresh_latch()
|
||||
limit = 20_000_000
|
||||
s = CreditsState(
|
||||
subscription_limit_micros=limit,
|
||||
subscription_limit_usd="20.00",
|
||||
subscription_micros=1, # limit - 1 → used_fraction < 1.0
|
||||
denominator_kind="subscription_cap",
|
||||
purchased_micros=5_000_000,
|
||||
purchased_usd="5.00",
|
||||
paid_access=True,
|
||||
)
|
||||
assert s.used_fraction is not None and s.used_fraction < 1.0
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
assert all(n.key != "credits.grant_spent" for n in to_show)
|
||||
assert "credits.grant_spent" not in to_clear
|
||||
|
||||
|
||||
# ── Scenario 4: grant_spent ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestGrantSpent:
|
||||
def _grant_state(self, purchased_micros: int = 12_340_000) -> CreditsState:
|
||||
return state_with_fraction(
|
||||
1.0,
|
||||
denominator_kind="subscription_cap",
|
||||
purchased_micros=purchased_micros,
|
||||
purchased_usd="12.34",
|
||||
)
|
||||
|
||||
def test_grant_spent_fires_on_first_obs(self):
|
||||
"""No crossing gate for grant_spent — fires immediately on first obs."""
|
||||
latch = fresh_latch()
|
||||
to_show, to_clear = evaluate_credits_notices(self._grant_state(), latch)
|
||||
keys = [n.key for n in to_show]
|
||||
assert "credits.grant_spent" in keys
|
||||
|
||||
def test_grant_spent_no_refire(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(self._grant_state(), latch)
|
||||
to_show, to_clear = evaluate_credits_notices(self._grant_state(), latch)
|
||||
assert all(n.key != "credits.grant_spent" for n in to_show)
|
||||
assert "credits.grant_spent" not in to_clear
|
||||
|
||||
def test_grant_spent_clears_when_purchased_zero(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(self._grant_state(), latch)
|
||||
# Now purchased → 0: grant_cond becomes False
|
||||
s_no_purchase = state_with_fraction(
|
||||
1.0,
|
||||
denominator_kind="subscription_cap",
|
||||
purchased_micros=0,
|
||||
purchased_usd="0.00",
|
||||
)
|
||||
to_show, to_clear = evaluate_credits_notices(s_no_purchase, latch)
|
||||
assert "credits.grant_spent" in to_clear
|
||||
assert all(n.key != "credits.grant_spent" for n in to_show)
|
||||
|
||||
|
||||
# ── Scenario 5: depleted + recovery ──────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDepleted:
|
||||
def test_depleted_fires_level_error_kind_sticky(self):
|
||||
latch = fresh_latch()
|
||||
s = CreditsState(paid_access=False)
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
depleted_notices = [n for n in to_show if n.key == "credits.depleted"]
|
||||
assert len(depleted_notices) == 1
|
||||
n = depleted_notices[0]
|
||||
assert n.level == "error"
|
||||
assert n.kind == CREDITS_NOTICE_KIND
|
||||
|
||||
def test_recovery_emits_clear_and_restored(self):
|
||||
latch = fresh_latch()
|
||||
# Fire depleted
|
||||
evaluate_credits_notices(CreditsState(paid_access=False), latch)
|
||||
# Now recovered
|
||||
to_show, to_clear = evaluate_credits_notices(CreditsState(paid_access=True), latch)
|
||||
assert "credits.depleted" in to_clear
|
||||
restored = [n for n in to_show if n.key == "credits.restored"]
|
||||
assert len(restored) == 1
|
||||
r = restored[0]
|
||||
assert r.level == "success"
|
||||
assert r.kind == "ttl"
|
||||
assert r.ttl_ms == CREDITS_RESTORED_TTL_MS
|
||||
|
||||
def test_depleted_refires_after_recovery(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(CreditsState(paid_access=False), latch)
|
||||
evaluate_credits_notices(CreditsState(paid_access=True), latch)
|
||||
# Goes depleted again
|
||||
to_show, to_clear = evaluate_credits_notices(CreditsState(paid_access=False), latch)
|
||||
keys = [n.key for n in to_show]
|
||||
assert "credits.depleted" in keys
|
||||
|
||||
|
||||
# ── Scenario 6: denominator none (uf is None) ────────────────────────────────
|
||||
|
||||
|
||||
class TestDenominatorNone:
|
||||
def test_no_warn90_when_uf_none(self):
|
||||
latch = fresh_latch()
|
||||
s = state_with_fraction(None)
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
assert all(n.key != "credits.usage" for n in to_show)
|
||||
assert "credits.usage" not in to_clear
|
||||
|
||||
def test_no_grant_spent_when_uf_none(self):
|
||||
latch = fresh_latch()
|
||||
s = CreditsState(
|
||||
subscription_limit_micros=None,
|
||||
denominator_kind="none",
|
||||
purchased_micros=5_000_000,
|
||||
purchased_usd="5.00",
|
||||
)
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
assert all(n.key != "credits.grant_spent" for n in to_show)
|
||||
|
||||
def test_warn90_clears_when_uf_becomes_none(self):
|
||||
"""If warn90 was active and uf becomes None, it should clear."""
|
||||
latch = fresh_latch()
|
||||
# Establish usage notice active: cross below → above
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
evaluate_credits_notices(state_with_fraction(0.95), latch)
|
||||
assert "credits.usage" in latch["active"]
|
||||
# Now uf becomes None (denominator changed to "none")
|
||||
s_none = state_with_fraction(None)
|
||||
to_show, to_clear = evaluate_credits_notices(s_none, latch)
|
||||
assert "credits.usage" in to_clear
|
||||
assert "credits.usage" not in latch["active"]
|
||||
|
||||
|
||||
# ── Scenario 7: copy / verbatim USD strings ──────────────────────────────────
|
||||
|
||||
|
||||
class TestNoticeCopy:
|
||||
def test_warn90_contains_verbatim_subscription_limit_usd(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
s = state_with_fraction(0.95, subscription_limit_usd="20.00")
|
||||
to_show, _ = evaluate_credits_notices(s, latch)
|
||||
warn_notice = next(n for n in to_show if n.key == "credits.usage")
|
||||
assert "$20.00" in warn_notice.text
|
||||
assert "cap" in warn_notice.text
|
||||
|
||||
def test_grant_spent_contains_verbatim_purchased_usd(self):
|
||||
latch = fresh_latch()
|
||||
s = state_with_fraction(
|
||||
1.0,
|
||||
denominator_kind="subscription_cap",
|
||||
purchased_micros=12_340_000,
|
||||
purchased_usd="12.34",
|
||||
)
|
||||
to_show, _ = evaluate_credits_notices(s, latch)
|
||||
grant_notice = next(n for n in to_show if n.key == "credits.grant_spent")
|
||||
assert "$12.34" in grant_notice.text
|
||||
assert "top-up left" in grant_notice.text
|
||||
|
||||
def test_depleted_mentions_usage_command(self):
|
||||
latch = fresh_latch()
|
||||
s = CreditsState(paid_access=False)
|
||||
to_show, _ = evaluate_credits_notices(s, latch)
|
||||
depleted_notice = next(n for n in to_show if n.key == "credits.depleted")
|
||||
assert "/usage" in depleted_notice.text
|
||||
|
||||
|
||||
# ── Scenario 8: severity order in a single call ──────────────────────────────
|
||||
|
||||
|
||||
class TestSeverityOrder:
|
||||
def test_multiple_new_notices_ordered_ascending_severity(self):
|
||||
"""warn90 < grant_spent < depleted in to_show when all fire in one call."""
|
||||
# Construct a state where all three conditions fire simultaneously
|
||||
# on first call (no latch state yet):
|
||||
# - warn90: uf >= 0.9 AND seen_below_90 must be True → won't fire fresh latch
|
||||
# So we pre-seed seen_below_90=True to allow warn90 to fire.
|
||||
latch = {"active": set(), "seen_below_90": True, "usage_band": None}
|
||||
|
||||
# Build state: subscription_cap, uf >= 1.0, purchased_micros > 0, NOT paid_access
|
||||
# warn90_cond: uf >= 0.9 ✓ (uf=1.0)
|
||||
# grant_cond: subscription_cap + uf >= 1.0 + purchased > 0 ✓
|
||||
# depleted_cond: not paid_access ✓
|
||||
s = CreditsState(
|
||||
subscription_limit_micros=20_000_000,
|
||||
subscription_limit_usd="20.00",
|
||||
subscription_micros=0, # uf = 1.0
|
||||
denominator_kind="subscription_cap",
|
||||
purchased_micros=5_000_000,
|
||||
purchased_usd="5.00",
|
||||
paid_access=False,
|
||||
)
|
||||
to_show, _ = evaluate_credits_notices(s, latch)
|
||||
keys = [n.key for n in to_show]
|
||||
assert "credits.usage" in keys
|
||||
assert "credits.grant_spent" in keys
|
||||
assert "credits.depleted" in keys
|
||||
# Ascending severity: warn90 before grant_spent before depleted
|
||||
assert keys.index("credits.usage") < keys.index("credits.grant_spent")
|
||||
assert keys.index("credits.grant_spent") < keys.index("credits.depleted")
|
||||
|
||||
|
||||
# ── Invariant: never fire + clear same key in one call ────────────────────────
|
||||
|
||||
|
||||
class TestNoFireAndClearSameKey:
|
||||
def test_usage_never_both_fired_and_cleared(self):
|
||||
latch = fresh_latch()
|
||||
# Run many state transitions; across each, assert no key is in both lists
|
||||
states = [
|
||||
state_with_fraction(0.10),
|
||||
state_with_fraction(0.95),
|
||||
state_with_fraction(0.10),
|
||||
state_with_fraction(0.95),
|
||||
state_with_fraction(None),
|
||||
]
|
||||
for s in states:
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
fired_keys = {n.key for n in to_show}
|
||||
cleared_keys = set(to_clear)
|
||||
overlap = fired_keys & cleared_keys
|
||||
assert not overlap, f"Key(s) both fired and cleared: {overlap}"
|
||||
|
||||
def test_depleted_never_both_fired_and_cleared(self):
|
||||
latch = fresh_latch()
|
||||
states = [
|
||||
CreditsState(paid_access=False),
|
||||
CreditsState(paid_access=True),
|
||||
CreditsState(paid_access=False),
|
||||
]
|
||||
for s in states:
|
||||
to_show, to_clear = evaluate_credits_notices(s, latch)
|
||||
fired_keys = {n.key for n in to_show}
|
||||
cleared_keys = set(to_clear)
|
||||
overlap = fired_keys & cleared_keys
|
||||
assert not overlap, f"Key(s) both fired and cleared: {overlap}"
|
||||
|
||||
|
||||
# ── Scenario 9: escalating usage bands (50 → 75 → 90) ────────────────────────
|
||||
|
||||
|
||||
class TestUsageBands:
|
||||
"""The usage notice shows the HIGHEST crossed band as a single escalating line."""
|
||||
|
||||
def _band_text(self, to_show):
|
||||
n = next((n for n in to_show if n.key == "credits.usage"), None)
|
||||
return n.text if n else None
|
||||
|
||||
def test_50_band_fires_info(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch) # prime
|
||||
to_show, _ = evaluate_credits_notices(state_with_fraction(0.55), latch)
|
||||
n = next(n for n in to_show if n.key == "credits.usage")
|
||||
assert "50%" in n.text and n.level == "info"
|
||||
assert latch["usage_band"] == 50
|
||||
|
||||
def test_75_band_fires_warn(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
to_show, _ = evaluate_credits_notices(state_with_fraction(0.80), latch)
|
||||
n = next(n for n in to_show if n.key == "credits.usage")
|
||||
assert "75%" in n.text and n.level == "warn"
|
||||
assert latch["usage_band"] == 75
|
||||
|
||||
def test_climb_replaces_band(self):
|
||||
"""Climbing 50→75→90 replaces the single line (clear old + show new)."""
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
# 55% → 50 band
|
||||
evaluate_credits_notices(state_with_fraction(0.55), latch)
|
||||
assert latch["usage_band"] == 50
|
||||
# 80% → climbs to 75, clearing the 50 line
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.80), latch)
|
||||
assert "credits.usage" in to_clear
|
||||
assert "75%" in self._band_text(to_show)
|
||||
assert latch["usage_band"] == 75
|
||||
# 95% → climbs to 90
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.95), latch)
|
||||
assert "credits.usage" in to_clear
|
||||
assert "90%" in self._band_text(to_show)
|
||||
assert latch["usage_band"] == 90
|
||||
|
||||
def test_step_down_on_recovery(self):
|
||||
"""Recovering steps the band back down, then clears below the lowest band."""
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
evaluate_credits_notices(state_with_fraction(0.95), latch)
|
||||
assert latch["usage_band"] == 90
|
||||
# drop to 80% → steps down to 75
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.80), latch)
|
||||
assert "credits.usage" in to_clear
|
||||
assert "75%" in self._band_text(to_show)
|
||||
# drop to 55% → steps down to 50
|
||||
to_show, _ = evaluate_credits_notices(state_with_fraction(0.55), latch)
|
||||
assert "50%" in self._band_text(to_show)
|
||||
# drop below 50% → clears entirely
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
assert "credits.usage" in to_clear
|
||||
assert latch["usage_band"] is None
|
||||
|
||||
def test_no_refire_same_band(self):
|
||||
latch = fresh_latch()
|
||||
evaluate_credits_notices(state_with_fraction(0.10), latch)
|
||||
evaluate_credits_notices(state_with_fraction(0.80), latch) # fires 75
|
||||
# still 80% → same band, no re-emit, no clear
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.80), latch)
|
||||
assert all(n.key != "credits.usage" for n in to_show)
|
||||
assert "credits.usage" not in to_clear
|
||||
|
||||
def test_exact_band_boundaries_inclusive(self):
|
||||
"""Thresholds are inclusive: exactly 0.50 / 0.75 / 0.90 land in their band."""
|
||||
for uf, want in [(0.50, 50), (0.75, 75), (0.90, 90)]:
|
||||
latch = fresh_latch()
|
||||
latch["seen_below_90"] = True # allow firing
|
||||
evaluate_credits_notices(state_with_fraction(uf), latch)
|
||||
assert latch["usage_band"] == want, (uf, latch["usage_band"])
|
||||
|
||||
def test_open_below_lowest_band_no_notice(self):
|
||||
latch = fresh_latch()
|
||||
to_show, to_clear = evaluate_credits_notices(state_with_fraction(0.30), latch)
|
||||
assert all(n.key != "credits.usage" for n in to_show)
|
||||
assert latch["usage_band"] is None
|
||||
@@ -0,0 +1,909 @@
|
||||
"""Tests for agent.credits_tracker — CreditsState + parse_credits_headers.
|
||||
|
||||
Covers the 9-state matrix plus validation edge cases. All header values
|
||||
arrive as STRINGS (the producer calls String(...) on every field).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import Optional
|
||||
import pytest
|
||||
|
||||
from agent.credits_tracker import CreditsState, parse_credits_headers
|
||||
|
||||
|
||||
# ── Helpers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def micros(dollars: float) -> str:
|
||||
"""Convert a dollar amount to a micros string for header fixtures."""
|
||||
return str(round(dollars * 1_000_000))
|
||||
|
||||
|
||||
# ── 9-State matrix fixtures ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _base_headers(**overrides) -> dict:
|
||||
"""Base headers present in every valid response."""
|
||||
h = {
|
||||
"x-nous-credits-version": "1",
|
||||
"x-nous-credits-remaining-micros": micros(0),
|
||||
"x-nous-credits-remaining-usd": "0.00",
|
||||
"x-nous-credits-subscription-micros": micros(0),
|
||||
"x-nous-credits-subscription-usd": "0.00",
|
||||
"x-nous-credits-rollover-micros": micros(0),
|
||||
"x-nous-credits-purchased-micros": micros(0),
|
||||
"x-nous-credits-purchased-usd": "0.00",
|
||||
"x-nous-tool-pool-micros": micros(0),
|
||||
"x-nous-tool-pool-gated-off": "false",
|
||||
"x-nous-credits-denominator-kind": "none",
|
||||
"x-nous-credits-paid-access": "true",
|
||||
"x-nous-credits-as-of-ms": "1717000000000",
|
||||
}
|
||||
h.update(overrides)
|
||||
return h
|
||||
|
||||
|
||||
# ── 9 STATES ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
HEALTHY_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(30.34),
|
||||
"x-nous-credits-remaining-usd": "30.34",
|
||||
"x-nous-credits-subscription-micros": micros(18.00),
|
||||
"x-nous-credits-subscription-usd": "18.00",
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-subscription-limit-usd": "20.00",
|
||||
"x-nous-credits-rollover-micros": micros(0),
|
||||
"x-nous-credits-purchased-micros": micros(12.34),
|
||||
"x-nous-credits-purchased-usd": "12.34",
|
||||
"x-nous-tool-pool-micros": micros(2.00),
|
||||
"x-nous-tool-pool-gated-off": "true",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
"x-nous-credits-paid-access": "true",
|
||||
}
|
||||
)
|
||||
|
||||
SUB_90PCT_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(2.00),
|
||||
"x-nous-credits-remaining-usd": "2.00",
|
||||
"x-nous-credits-subscription-micros": micros(2.00),
|
||||
"x-nous-credits-subscription-usd": "2.00",
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-subscription-limit-usd": "20.00",
|
||||
"x-nous-credits-purchased-micros": micros(0),
|
||||
"x-nous-credits-purchased-usd": "0.00",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
"x-nous-credits-paid-access": "true",
|
||||
}
|
||||
)
|
||||
|
||||
GRANT_EXHAUSTED_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(12.34),
|
||||
"x-nous-credits-remaining-usd": "12.34",
|
||||
"x-nous-credits-subscription-micros": micros(0),
|
||||
"x-nous-credits-subscription-usd": "0.00",
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-subscription-limit-usd": "20.00",
|
||||
"x-nous-credits-purchased-micros": micros(12.34),
|
||||
"x-nous-credits-purchased-usd": "12.34",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
"x-nous-credits-paid-access": "true",
|
||||
}
|
||||
)
|
||||
|
||||
PURCHASED_ONLY_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(30.00),
|
||||
"x-nous-credits-remaining-usd": "30.00",
|
||||
"x-nous-credits-subscription-micros": micros(0),
|
||||
"x-nous-credits-subscription-usd": "0.00",
|
||||
"x-nous-credits-purchased-micros": micros(30.00),
|
||||
"x-nous-credits-purchased-usd": "30.00",
|
||||
"x-nous-credits-denominator-kind": "none",
|
||||
"x-nous-credits-paid-access": "true",
|
||||
# No limit pair — denominator_kind=none
|
||||
}
|
||||
)
|
||||
|
||||
TOOL_POOL_FREE_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(0.05),
|
||||
"x-nous-credits-remaining-usd": "0.05",
|
||||
"x-nous-tool-pool-micros": micros(0.05),
|
||||
"x-nous-tool-pool-gated-off": "false",
|
||||
"x-nous-credits-paid-access": "true",
|
||||
}
|
||||
)
|
||||
|
||||
DEPLETED_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(0),
|
||||
"x-nous-credits-remaining-usd": "0.00",
|
||||
"x-nous-credits-subscription-micros": micros(0),
|
||||
"x-nous-credits-subscription-usd": "0.00",
|
||||
"x-nous-credits-purchased-micros": micros(0),
|
||||
"x-nous-credits-purchased-usd": "0.00",
|
||||
"x-nous-credits-paid-access": "false",
|
||||
"x-nous-credits-disabled-reason": "out_of_credits",
|
||||
}
|
||||
)
|
||||
|
||||
DEBT_HEADERS = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-remaining-micros": micros(0),
|
||||
"x-nous-credits-remaining-usd": "0.00",
|
||||
"x-nous-credits-subscription-micros": str(-5_000_000),
|
||||
"x-nous-credits-subscription-usd": "-5.00",
|
||||
"x-nous-credits-purchased-micros": micros(0),
|
||||
"x-nous-credits-purchased-usd": "0.00",
|
||||
"x-nous-credits-paid-access": "false",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── State 1: healthy ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestHealthyState:
|
||||
def test_parses_successfully(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state is not None
|
||||
|
||||
def test_from_header_set(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.from_header is True
|
||||
|
||||
def test_captured_at_set(self):
|
||||
before = time.time()
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
after = time.time()
|
||||
assert before <= state.captured_at <= after
|
||||
|
||||
def test_remaining_fields(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.remaining_micros == round(30.34 * 1_000_000)
|
||||
assert state.remaining_usd == "30.34"
|
||||
|
||||
def test_subscription_fields(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.subscription_micros == round(18.00 * 1_000_000)
|
||||
assert state.subscription_usd == "18.00"
|
||||
assert state.subscription_limit_micros == round(20.00 * 1_000_000)
|
||||
assert state.subscription_limit_usd == "20.00"
|
||||
|
||||
def test_rollover_and_purchased(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.rollover_micros == 0
|
||||
assert state.purchased_micros == round(12.34 * 1_000_000)
|
||||
assert state.purchased_usd == "12.34"
|
||||
|
||||
def test_tool_pool(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.tool_pool_micros == round(2.00 * 1_000_000)
|
||||
assert state.tool_pool_gated_off is True
|
||||
|
||||
def test_denominator_and_access(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.denominator_kind == "subscription_cap"
|
||||
assert state.paid_access is True
|
||||
assert state.disabled_reason is None
|
||||
|
||||
def test_used_fraction(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
# (20.00 - 18.00) / 20.00 = 0.10
|
||||
assert state.used_fraction == pytest.approx(0.10)
|
||||
|
||||
def test_has_data(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.has_data is True
|
||||
|
||||
def test_not_depleted(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
assert state.depleted is False
|
||||
|
||||
def test_age_seconds_reasonable(self):
|
||||
state = parse_credits_headers(HEALTHY_HEADERS)
|
||||
# Should be very small — just parsed
|
||||
assert 0 <= state.age_seconds < 5
|
||||
|
||||
|
||||
# ── State 2: sub_90pct ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSub90Pct:
|
||||
def test_parses_successfully(self):
|
||||
state = parse_credits_headers(SUB_90PCT_HEADERS)
|
||||
assert state is not None
|
||||
|
||||
def test_used_fraction_90pct(self):
|
||||
state = parse_credits_headers(SUB_90PCT_HEADERS)
|
||||
# (20.00 - 2.00) / 20.00 = 0.90
|
||||
assert state.used_fraction == pytest.approx(0.90)
|
||||
|
||||
def test_paid_access(self):
|
||||
state = parse_credits_headers(SUB_90PCT_HEADERS)
|
||||
assert state.paid_access is True
|
||||
assert state.depleted is False
|
||||
|
||||
|
||||
# ── State 3: grant_exhausted ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestGrantExhausted:
|
||||
def test_used_fraction_100pct(self):
|
||||
state = parse_credits_headers(GRANT_EXHAUSTED_HEADERS)
|
||||
assert state is not None
|
||||
# subscription_micros=0, limit=20.00 → (20-0)/20 = 1.0
|
||||
assert state.used_fraction == pytest.approx(1.0)
|
||||
|
||||
def test_paid_access_still_true(self):
|
||||
state = parse_credits_headers(GRANT_EXHAUSTED_HEADERS)
|
||||
assert state.paid_access is True
|
||||
assert state.depleted is False
|
||||
|
||||
|
||||
# ── State 4: purchased_only ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPurchasedOnly:
|
||||
def test_parses_successfully(self):
|
||||
state = parse_credits_headers(PURCHASED_ONLY_HEADERS)
|
||||
assert state is not None
|
||||
|
||||
def test_denominator_kind_none(self):
|
||||
state = parse_credits_headers(PURCHASED_ONLY_HEADERS)
|
||||
assert state.denominator_kind == "none"
|
||||
|
||||
def test_used_fraction_is_none_no_limit(self):
|
||||
state = parse_credits_headers(PURCHASED_ONLY_HEADERS)
|
||||
# No subscription_limit_micros → used_fraction is None
|
||||
assert state.used_fraction is None
|
||||
|
||||
def test_no_limit_pair(self):
|
||||
state = parse_credits_headers(PURCHASED_ONLY_HEADERS)
|
||||
assert state.subscription_limit_micros is None
|
||||
assert state.subscription_limit_usd is None
|
||||
|
||||
|
||||
# ── State 5: tool_pool_free ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestToolPoolFree:
|
||||
def test_parses_successfully(self):
|
||||
state = parse_credits_headers(TOOL_POOL_FREE_HEADERS)
|
||||
assert state is not None
|
||||
|
||||
def test_tool_pool_gated_off_false(self):
|
||||
state = parse_credits_headers(TOOL_POOL_FREE_HEADERS)
|
||||
assert state.tool_pool_gated_off is False
|
||||
|
||||
def test_tool_pool_micros(self):
|
||||
state = parse_credits_headers(TOOL_POOL_FREE_HEADERS)
|
||||
assert state.tool_pool_micros == round(0.05 * 1_000_000)
|
||||
|
||||
def test_paid_access(self):
|
||||
state = parse_credits_headers(TOOL_POOL_FREE_HEADERS)
|
||||
assert state.paid_access is True
|
||||
|
||||
|
||||
# ── State 6: depleted ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDepleted:
|
||||
def test_parses_successfully(self):
|
||||
state = parse_credits_headers(DEPLETED_HEADERS)
|
||||
assert state is not None
|
||||
|
||||
def test_paid_access_false(self):
|
||||
state = parse_credits_headers(DEPLETED_HEADERS)
|
||||
assert state.paid_access is False
|
||||
|
||||
def test_depleted_true(self):
|
||||
state = parse_credits_headers(DEPLETED_HEADERS)
|
||||
assert state.depleted is True
|
||||
|
||||
def test_disabled_reason(self):
|
||||
state = parse_credits_headers(DEPLETED_HEADERS)
|
||||
assert state.disabled_reason == "out_of_credits"
|
||||
|
||||
def test_remaining_zero(self):
|
||||
state = parse_credits_headers(DEPLETED_HEADERS)
|
||||
assert state.remaining_micros == 0
|
||||
|
||||
|
||||
# ── State 7: debt ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDebt:
|
||||
def test_parses_successfully(self):
|
||||
# Negative subscription_micros should NOT cause the parse to fail
|
||||
state = parse_credits_headers(DEBT_HEADERS)
|
||||
assert state is not None
|
||||
|
||||
def test_negative_subscription_accepted(self):
|
||||
state = parse_credits_headers(DEBT_HEADERS)
|
||||
assert state.subscription_micros == -5_000_000
|
||||
|
||||
def test_negative_subscription_usd_accepted(self):
|
||||
state = parse_credits_headers(DEBT_HEADERS)
|
||||
assert state.subscription_usd == "-5.00"
|
||||
|
||||
def test_paid_access_false(self):
|
||||
state = parse_credits_headers(DEBT_HEADERS)
|
||||
assert state.paid_access is False
|
||||
assert state.depleted is True
|
||||
|
||||
|
||||
# ── State 8: missing ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMissing:
|
||||
def test_no_credits_headers_returns_none(self):
|
||||
state = parse_credits_headers({})
|
||||
assert state is None
|
||||
|
||||
def test_completely_empty_dict(self):
|
||||
assert parse_credits_headers({}) is None
|
||||
|
||||
|
||||
# ── State 9: no_org ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestNoOrg:
|
||||
def test_irrelevant_headers_return_none(self):
|
||||
headers = {
|
||||
"content-type": "application/json",
|
||||
"x-request-id": "abc123",
|
||||
"server": "nginx",
|
||||
}
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is None
|
||||
|
||||
def test_api_key_path_no_org_returns_none(self):
|
||||
# Headers that might appear on an api-key path with no org
|
||||
headers = {
|
||||
"content-type": "application/json",
|
||||
"authorization": "Bearer sk-test",
|
||||
}
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
|
||||
# ── Version validation ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVersionValidation:
|
||||
def test_version_string_1_parses(self):
|
||||
headers = _base_headers(**{"x-nous-credits-version": "1"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.version == 1
|
||||
|
||||
def test_version_2_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-version": "2"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is None
|
||||
|
||||
def test_version_absent_returns_none(self):
|
||||
headers = {k: v for k, v in _base_headers().items() if k != "x-nous-credits-version"}
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is None
|
||||
|
||||
def test_version_greater_than_1_warns_once(self, caplog):
|
||||
"""Version > 1 must log a warning, and ONLY ONCE across multiple calls."""
|
||||
import agent.credits_tracker as ct
|
||||
|
||||
original = ct._version_warning_emitted
|
||||
try:
|
||||
# Reset the warn-once latch so this test starts clean regardless of order
|
||||
ct._version_warning_emitted = False
|
||||
|
||||
headers = _base_headers(**{"x-nous-credits-version": "3"})
|
||||
with caplog.at_level(logging.WARNING, logger="agent.credits_tracker"):
|
||||
parse_credits_headers(headers)
|
||||
parse_credits_headers(headers)
|
||||
parse_credits_headers(headers)
|
||||
|
||||
warning_records = [r for r in caplog.records if "unsupported" in r.message.lower() or "version" in r.message.lower()]
|
||||
assert len(warning_records) == 1, (
|
||||
f"Expected exactly 1 version warning, got {len(warning_records)}: {[r.message for r in warning_records]}"
|
||||
)
|
||||
finally:
|
||||
ct._version_warning_emitted = original
|
||||
|
||||
def test_version_0_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-version": "0"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_version_non_int_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-version": "abc"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
|
||||
# ── Bool-string trap ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestBoolStringTrap:
|
||||
"""Explicit tests for the bool("false") == True trap."""
|
||||
|
||||
def test_paid_access_string_false_means_depleted(self):
|
||||
"""paid_access='false' must yield paid_access=False — NOT True."""
|
||||
headers = _base_headers(**{"x-nous-credits-paid-access": "false"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.paid_access is False
|
||||
assert state.depleted is True
|
||||
|
||||
def test_paid_access_string_true_means_not_depleted(self):
|
||||
headers = _base_headers(**{"x-nous-credits-paid-access": "true"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.paid_access is True
|
||||
assert state.depleted is False
|
||||
|
||||
def test_paid_access_case_insensitive_FALSE(self):
|
||||
headers = _base_headers(**{"x-nous-credits-paid-access": "FALSE"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.paid_access is False
|
||||
|
||||
def test_paid_access_case_insensitive_True(self):
|
||||
headers = _base_headers(**{"x-nous-credits-paid-access": "True"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.paid_access is True
|
||||
|
||||
def test_tool_pool_gated_off_false(self):
|
||||
headers = _base_headers(**{"x-nous-tool-pool-gated-off": "false"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.tool_pool_gated_off is False
|
||||
|
||||
def test_tool_pool_gated_off_true(self):
|
||||
headers = _base_headers(**{"x-nous-tool-pool-gated-off": "true"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.tool_pool_gated_off is True
|
||||
|
||||
|
||||
# ── Tool-pool optional headers ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestToolPoolOptional:
|
||||
"""x-nous-tool-pool-* headers are optional; absent → defaults; present-but-malformed → miss."""
|
||||
|
||||
def _no_tool_pool_headers(self) -> dict:
|
||||
"""Base headers with BOTH tool-pool headers removed."""
|
||||
h = _base_headers()
|
||||
h.pop("x-nous-tool-pool-micros", None)
|
||||
h.pop("x-nous-tool-pool-gated-off", None)
|
||||
return h
|
||||
|
||||
def test_absent_tool_pool_headers_parse_succeeds(self):
|
||||
"""Valid credits headers with no x-nous-tool-pool-* → parse succeeds."""
|
||||
state = parse_credits_headers(self._no_tool_pool_headers())
|
||||
assert state is not None
|
||||
|
||||
def test_absent_tool_pool_micros_defaults_to_zero(self):
|
||||
state = parse_credits_headers(self._no_tool_pool_headers())
|
||||
assert state.tool_pool_micros == 0
|
||||
|
||||
def test_absent_tool_pool_gated_off_defaults_to_false(self):
|
||||
state = parse_credits_headers(self._no_tool_pool_headers())
|
||||
assert state.tool_pool_gated_off is False
|
||||
|
||||
def test_present_malformed_tool_pool_micros_returns_none(self):
|
||||
"""x-nous-tool-pool-micros present but non-int → parse miss (returns None)."""
|
||||
headers = _base_headers(**{"x-nous-tool-pool-micros": "not-a-number"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_present_negative_tool_pool_micros_returns_none(self):
|
||||
"""x-nous-tool-pool-micros present but negative → parse miss (returns None)."""
|
||||
headers = _base_headers(**{"x-nous-tool-pool-micros": "-1000"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_only_tool_pool_micros_absent_still_succeeds(self):
|
||||
"""Only micros absent (gated-off still present) → tool_pool_micros = 0, parse succeeds."""
|
||||
h = _base_headers()
|
||||
h.pop("x-nous-tool-pool-micros", None)
|
||||
state = parse_credits_headers(h)
|
||||
assert state is not None
|
||||
assert state.tool_pool_micros == 0
|
||||
|
||||
|
||||
# ── Half-pair subscription limit ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestHalfPairLimit:
|
||||
def test_only_limit_micros_present_both_absent(self):
|
||||
"""Only -micros present → both None, parse SUCCEEDS."""
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.subscription_limit_micros is None
|
||||
assert state.subscription_limit_usd is None
|
||||
|
||||
def test_only_limit_usd_present_both_absent(self):
|
||||
"""Only -usd present → both None, parse SUCCEEDS."""
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-limit-usd": "20.00",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.subscription_limit_micros is None
|
||||
assert state.subscription_limit_usd is None
|
||||
|
||||
def test_half_pair_used_fraction_is_none(self):
|
||||
"""With no limit pair, used_fraction is None regardless of denominator_kind."""
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.used_fraction is None
|
||||
|
||||
def test_full_pair_present_parsed_correctly(self):
|
||||
"""Both present → both populated, used_fraction computable."""
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-micros": micros(10.00),
|
||||
"x-nous-credits-subscription-usd": "10.00",
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-subscription-limit-usd": "20.00",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.subscription_limit_micros == round(20.00 * 1_000_000)
|
||||
assert state.subscription_limit_usd == "20.00"
|
||||
assert state.used_fraction == pytest.approx(0.50)
|
||||
|
||||
|
||||
# ── Negative value validation ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestNegativeValues:
|
||||
def test_negative_remaining_micros_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-micros": "-1000"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_negative_purchased_micros_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-purchased-micros": "-500"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_negative_rollover_micros_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-rollover-micros": "-100"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_negative_limit_micros_returns_none(self):
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-limit-micros": "-1000",
|
||||
"x-nous-credits-subscription-limit-usd": "-0.00",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
}
|
||||
)
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_negative_subscription_accepted(self):
|
||||
"""subscription_micros is the ONLY field allowed to be negative."""
|
||||
headers = _base_headers(**{"x-nous-credits-subscription-micros": "-5000000",
|
||||
"x-nous-credits-subscription-usd": "-5.00"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.subscription_micros == -5_000_000
|
||||
|
||||
|
||||
# ── USD format validation ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestUsdValidation:
|
||||
def test_valid_usd_format(self):
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-usd": "18.00"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.remaining_usd == "18.00"
|
||||
|
||||
def test_usd_one_decimal_returns_none(self):
|
||||
"""'18.0' does not match ^-?\d+\.\d{2}$"""
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-usd": "18.0"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_usd_no_decimal_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-usd": "18"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_usd_with_dollar_sign_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-usd": "$18.00"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_usd_with_comma_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-usd": "1,800.00"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_usd_negative_valid(self):
|
||||
"""Negative USD string should parse (e.g. subscription debt)."""
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-micros": "-5000000",
|
||||
"x-nous-credits-subscription-usd": "-5.00",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.subscription_usd == "-5.00"
|
||||
|
||||
|
||||
# ── Non-int micros validation ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMicrosValidation:
|
||||
def test_non_int_micros_string_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-micros": "abc"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_float_string_micros_returns_none(self):
|
||||
"""'1.5' is not an integer string — should fail validation."""
|
||||
headers = _base_headers(**{"x-nous-credits-remaining-micros": "1.5"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_non_int_purchased_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-purchased-micros": "abc"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
|
||||
# ── as_of_ms validation ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAsOfMs:
|
||||
def test_junk_as_of_ms_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-as-of-ms": "not-a-timestamp"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
def test_valid_as_of_ms(self):
|
||||
headers = _base_headers(**{"x-nous-credits-as-of-ms": "1717000000000"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.as_of_ms == 1717000000000
|
||||
|
||||
|
||||
# ── denominator_kind validation ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDenominatorKind:
|
||||
def test_subscription_cap_valid(self):
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
"x-nous-credits-subscription-limit-micros": micros(20.00),
|
||||
"x-nous-credits-subscription-limit-usd": "20.00",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.denominator_kind == "subscription_cap"
|
||||
|
||||
def test_none_valid(self):
|
||||
headers = _base_headers(**{"x-nous-credits-denominator-kind": "none"})
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.denominator_kind == "none"
|
||||
|
||||
def test_invalid_denominator_kind_returns_none(self):
|
||||
headers = _base_headers(**{"x-nous-credits-denominator-kind": "invalid_kind"})
|
||||
assert parse_credits_headers(headers) is None
|
||||
|
||||
|
||||
# ── Zero-division guard ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestZeroDivisionGuard:
|
||||
def test_subscription_limit_zero_used_fraction_is_none(self):
|
||||
"""subscription_limit_micros='0' + subscription_cap → used_fraction is None (no ZeroDivisionError)."""
|
||||
headers = _base_headers(
|
||||
**{
|
||||
"x-nous-credits-subscription-limit-micros": "0",
|
||||
"x-nous-credits-subscription-limit-usd": "0.00",
|
||||
"x-nous-credits-denominator-kind": "subscription_cap",
|
||||
}
|
||||
)
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
# limit == 0, so used_fraction must be None (guard prevents division)
|
||||
assert state.used_fraction is None
|
||||
|
||||
|
||||
# ── Unknown headers ignored ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestUnknownHeaders:
|
||||
def test_unknown_extra_header_ignored(self):
|
||||
headers = {
|
||||
**_base_headers(),
|
||||
"x-nous-credits-future-field": "some-value",
|
||||
"x-request-id": "abc123",
|
||||
}
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
|
||||
def test_mixed_with_other_providers_headers(self):
|
||||
headers = {
|
||||
**_base_headers(),
|
||||
"x-ratelimit-limit-requests": "800",
|
||||
"content-type": "application/json",
|
||||
}
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
|
||||
|
||||
# ── Header normalization ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestHeaderNormalization:
|
||||
def test_uppercase_headers_parsed(self):
|
||||
headers = {k.upper(): v for k, v in _base_headers().items()}
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
|
||||
def test_mixed_case_headers_parsed(self):
|
||||
headers = {
|
||||
"X-Nous-Credits-Version": "1",
|
||||
"X-Nous-Credits-Remaining-Micros": micros(5.00),
|
||||
"X-Nous-Credits-Remaining-Usd": "5.00",
|
||||
"X-Nous-Credits-Subscription-Micros": micros(5.00),
|
||||
"X-Nous-Credits-Subscription-Usd": "5.00",
|
||||
"X-Nous-Credits-Rollover-Micros": "0",
|
||||
"X-Nous-Credits-Purchased-Micros": "0",
|
||||
"X-Nous-Credits-Purchased-Usd": "0.00",
|
||||
"X-Nous-Tool-Pool-Micros": "0",
|
||||
"X-Nous-Tool-Pool-Gated-Off": "false",
|
||||
"X-Nous-Credits-Denominator-Kind": "none",
|
||||
"X-Nous-Credits-Paid-Access": "true",
|
||||
"X-Nous-Credits-As-Of-Ms": "1717000000000",
|
||||
}
|
||||
state = parse_credits_headers(headers)
|
||||
assert state is not None
|
||||
assert state.remaining_micros == round(5.00 * 1_000_000)
|
||||
|
||||
|
||||
# ── CreditsState dataclass defaults ──────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCreditsStateDefaults:
|
||||
def test_default_state(self):
|
||||
state = CreditsState()
|
||||
assert state.version == 0
|
||||
assert state.remaining_micros == 0
|
||||
assert state.remaining_usd == ""
|
||||
assert state.subscription_micros == 0
|
||||
assert state.subscription_usd == ""
|
||||
assert state.subscription_limit_micros is None
|
||||
assert state.subscription_limit_usd is None
|
||||
assert state.rollover_micros == 0
|
||||
assert state.purchased_micros == 0
|
||||
assert state.purchased_usd == ""
|
||||
assert state.tool_pool_micros == 0
|
||||
assert state.tool_pool_gated_off is False
|
||||
assert state.denominator_kind == "none"
|
||||
assert state.paid_access is True
|
||||
assert state.disabled_reason is None
|
||||
assert state.as_of_ms == 0
|
||||
assert state.captured_at == 0.0
|
||||
assert state.from_header is False
|
||||
|
||||
def test_has_data_false_when_no_captured_at(self):
|
||||
state = CreditsState()
|
||||
assert state.has_data is False
|
||||
|
||||
def test_age_seconds_inf_when_no_data(self):
|
||||
state = CreditsState()
|
||||
assert state.age_seconds == float("inf")
|
||||
|
||||
def test_depleted_false_by_default(self):
|
||||
state = CreditsState()
|
||||
assert state.depleted is False
|
||||
|
||||
def test_used_fraction_none_by_default(self):
|
||||
state = CreditsState()
|
||||
assert state.used_fraction is None
|
||||
|
||||
|
||||
# ── depleted property ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestDepletedProperty:
|
||||
def test_depleted_equals_not_paid_access(self):
|
||||
"""depleted must be exactly `not paid_access`, never `remaining==0`."""
|
||||
state = CreditsState(paid_access=False, remaining_micros=0, captured_at=time.time())
|
||||
assert state.depleted is True
|
||||
|
||||
def test_not_depleted_when_paid_access_true(self):
|
||||
state = CreditsState(paid_access=True, remaining_micros=0, captured_at=time.time())
|
||||
# remaining==0 but paid_access is True → NOT depleted
|
||||
assert state.depleted is False
|
||||
|
||||
def test_depleted_independent_of_remaining(self):
|
||||
"""Even with remaining > 0, if paid_access is False, depleted is True."""
|
||||
state = CreditsState(paid_access=False, remaining_micros=1_000_000, captured_at=time.time())
|
||||
assert state.depleted is True
|
||||
|
||||
|
||||
# ── used_fraction edge cases ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestUsedFraction:
|
||||
def test_none_without_limit(self):
|
||||
state = CreditsState(
|
||||
denominator_kind="subscription_cap",
|
||||
subscription_limit_micros=None,
|
||||
captured_at=time.time(),
|
||||
)
|
||||
assert state.used_fraction is None
|
||||
|
||||
def test_none_when_limit_zero(self):
|
||||
state = CreditsState(
|
||||
denominator_kind="subscription_cap",
|
||||
subscription_limit_micros=0,
|
||||
subscription_micros=0,
|
||||
captured_at=time.time(),
|
||||
)
|
||||
assert state.used_fraction is None
|
||||
|
||||
def test_clamped_at_zero(self):
|
||||
"""If subscription_micros > limit (over-credited), fraction clamps to 0."""
|
||||
state = CreditsState(
|
||||
denominator_kind="subscription_cap",
|
||||
subscription_limit_micros=10_000_000,
|
||||
subscription_micros=15_000_000, # more than limit
|
||||
captured_at=time.time(),
|
||||
)
|
||||
assert state.used_fraction == pytest.approx(0.0)
|
||||
|
||||
def test_clamped_at_one(self):
|
||||
"""If subscription_micros is very negative (debt), fraction clamps to 1.0."""
|
||||
state = CreditsState(
|
||||
denominator_kind="subscription_cap",
|
||||
subscription_limit_micros=10_000_000,
|
||||
subscription_micros=-5_000_000, # deep debt
|
||||
captured_at=time.time(),
|
||||
)
|
||||
assert state.used_fraction == pytest.approx(1.0)
|
||||
|
||||
def test_guarded_by_limit_field_not_denominator(self):
|
||||
"""used_fraction depends on subscription_limit_micros being truthy, not denominator_kind."""
|
||||
# limit present but denominator_kind="none" — spec says guard on LIMIT FIELD
|
||||
state = CreditsState(
|
||||
denominator_kind="none",
|
||||
subscription_limit_micros=20_000_000,
|
||||
subscription_micros=10_000_000,
|
||||
captured_at=time.time(),
|
||||
)
|
||||
# With limit_micros set, fraction should be computable regardless of denominator_kind
|
||||
assert state.used_fraction == pytest.approx(0.50)
|
||||
|
||||
def test_none_when_denominator_cap_but_no_limit(self):
|
||||
"""denominator_kind=subscription_cap but no limit pair → None."""
|
||||
state = CreditsState(
|
||||
denominator_kind="subscription_cap",
|
||||
subscription_limit_micros=None,
|
||||
subscription_micros=5_000_000,
|
||||
captured_at=time.time(),
|
||||
)
|
||||
assert state.used_fraction is None
|
||||
@@ -0,0 +1,150 @@
|
||||
"""Tests for the Nous-credits subscription % gauge in build_nous_credits_snapshot.
|
||||
|
||||
Covers the monthly_credits denominator path added when the portal /api/oauth/account
|
||||
subscription block began carrying `monthly_credits`. Magnitudes-only fallback, clamp,
|
||||
and the non-finite / rollover guards (surfaced by adversarial review) are all asserted.
|
||||
"""
|
||||
from hermes_cli.nous_account import (
|
||||
NousPortalAccountInfo,
|
||||
NousPaidServiceAccessInfo,
|
||||
NousPortalSubscriptionInfo,
|
||||
_subscription_from_payload,
|
||||
)
|
||||
from agent.account_usage import build_nous_credits_snapshot, render_account_usage_lines
|
||||
|
||||
|
||||
def _acct(**kwargs):
|
||||
kwargs.setdefault("logged_in", True)
|
||||
kwargs.setdefault("source", "account_api")
|
||||
kwargs.setdefault("fresh", True)
|
||||
kwargs.setdefault("portal_base_url", "https://portal.nousresearch.com")
|
||||
return NousPortalAccountInfo(**kwargs)
|
||||
|
||||
|
||||
def _window(snap):
|
||||
return snap.windows[0] if (snap and snap.windows) else None
|
||||
|
||||
|
||||
def test_parser_captures_monthly_credits():
|
||||
sub = _subscription_from_payload({
|
||||
"plan": "Ultra", "tier": 14, "monthly_charge": 200, "monthly_credits": 220,
|
||||
"current_period_end": "2026-06-28T05:21:54.000Z",
|
||||
"credits_remaining": 219.27341839, "rollover_credits": 0,
|
||||
})
|
||||
assert sub.monthly_credits == 220
|
||||
assert abs(sub.credits_remaining - 219.27341839) < 1e-6
|
||||
|
||||
|
||||
def test_parser_monthly_credits_absent_is_none():
|
||||
sub = _subscription_from_payload({"plan": "Ultra", "credits_remaining": 10.0})
|
||||
assert sub.monthly_credits is None
|
||||
|
||||
|
||||
def test_gauge_present_with_monthly_credits():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(
|
||||
plan="Ultra", monthly_credits=220, credits_remaining=219.27341839,
|
||||
current_period_end="2026-06-28"),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(
|
||||
subscription_credits_remaining=219.27, total_usable_credits=219.27),
|
||||
))
|
||||
w = _window(snap)
|
||||
assert w is not None and w.label == "Subscription"
|
||||
assert abs(w.used_percent - (220 - 219.27341839) / 220 * 100) < 1e-9
|
||||
blob = "\n".join(render_account_usage_lines(snap))
|
||||
assert "% used" in blob or "% remaining" in blob
|
||||
assert "of $220.00 left" in blob
|
||||
|
||||
|
||||
def test_gauge_90pct():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=220, credits_remaining=22.0),
|
||||
))
|
||||
assert abs(_window(snap).used_percent - 90.0) < 1e-9
|
||||
|
||||
|
||||
def test_gauge_debt_clamps_to_100():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=False,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=220, credits_remaining=-5.0),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(subscription_credits_remaining=-5.0),
|
||||
))
|
||||
assert _window(snap).used_percent == 100.0
|
||||
|
||||
|
||||
def test_gauge_at_cap_is_zero_used():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=220, credits_remaining=220.0),
|
||||
))
|
||||
assert _window(snap).used_percent == 0.0
|
||||
|
||||
|
||||
def test_no_monthly_credits_falls_back_to_magnitudes():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(plan="Ultra", credits_remaining=-0.79),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(purchased_credits_remaining=991.96),
|
||||
))
|
||||
assert _window(snap) is None
|
||||
blob = "\n".join(render_account_usage_lines(snap))
|
||||
assert "%" not in blob
|
||||
assert "Top-up credits: $991.96" in blob
|
||||
|
||||
|
||||
def test_nan_remaining_no_window_no_nan_string():
|
||||
"""json.loads parses bare NaN by default; isinstance(nan, float) is True.
|
||||
The gauge must reject it rather than render '$nan' + a false 100% used."""
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=220, credits_remaining=float("nan")),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(purchased_credits_remaining=5.0),
|
||||
))
|
||||
assert _window(snap) is None
|
||||
assert "$nan" not in "\n".join(render_account_usage_lines(snap)).lower()
|
||||
|
||||
|
||||
def test_inf_cap_no_window():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=float("inf"), credits_remaining=10.0),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(purchased_credits_remaining=5.0),
|
||||
))
|
||||
assert _window(snap) is None
|
||||
|
||||
|
||||
def test_rollover_balance_exceeds_cap_no_window():
|
||||
"""remaining > cap (rollover spanning the period) makes monthly_credits a
|
||||
nonsensical denominator → suppress the gauge, keep magnitudes."""
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=220, credits_remaining=300, rollover_credits=80),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(subscription_credits_remaining=300.0),
|
||||
))
|
||||
assert _window(snap) is None
|
||||
assert "of $220.00 left" not in "\n".join(render_account_usage_lines(snap))
|
||||
|
||||
|
||||
def test_bool_monthly_credits_no_window():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=True, credits_remaining=1.0),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(purchased_credits_remaining=5.0),
|
||||
))
|
||||
assert _window(snap) is None
|
||||
|
||||
|
||||
def test_zero_monthly_credits_no_divzero():
|
||||
snap = build_nous_credits_snapshot(_acct(
|
||||
paid_service_access=True,
|
||||
subscription=NousPortalSubscriptionInfo(monthly_credits=0, credits_remaining=0.0),
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(purchased_credits_remaining=5.0),
|
||||
))
|
||||
assert _window(snap) is None
|
||||
|
||||
|
||||
def test_failopen_none_and_logged_out():
|
||||
assert build_nous_credits_snapshot(None) is None
|
||||
assert build_nous_credits_snapshot(_acct(logged_in=False)) is None
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Tests for build_nous_credits_snapshot (L6-A, magnitudes-only)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from agent.account_usage import build_nous_credits_snapshot
|
||||
from hermes_cli.nous_account import (
|
||||
NousPaidServiceAccessInfo,
|
||||
NousPortalAccountInfo,
|
||||
NousPortalSubscriptionInfo,
|
||||
)
|
||||
|
||||
|
||||
def _account(**kwargs) -> NousPortalAccountInfo:
|
||||
kwargs.setdefault("logged_in", True)
|
||||
kwargs.setdefault("source", "account_api")
|
||||
kwargs.setdefault("fresh", True)
|
||||
return NousPortalAccountInfo(**kwargs)
|
||||
|
||||
|
||||
def _all_lines(snapshot) -> list[str]:
|
||||
return list(snapshot.details)
|
||||
|
||||
|
||||
def test_healthy():
|
||||
info = _account(
|
||||
paid_service_access=True,
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(
|
||||
subscription_credits_remaining=18.0,
|
||||
purchased_credits_remaining=12.34,
|
||||
total_usable_credits=30.34,
|
||||
),
|
||||
subscription=NousPortalSubscriptionInfo(
|
||||
plan="Pro",
|
||||
current_period_end="2026-07-01",
|
||||
),
|
||||
)
|
||||
snap = build_nous_credits_snapshot(info)
|
||||
assert snap is not None
|
||||
assert snap.available is True
|
||||
assert snap.plan == "Pro"
|
||||
assert snap.provider == "nous"
|
||||
assert snap.title == "Nous credits"
|
||||
blob = "\n".join(_all_lines(snap))
|
||||
assert "$18.00" in blob
|
||||
assert "$12.34" in blob
|
||||
assert "$30.34" in blob
|
||||
assert "Renews: 2026-07-01" in blob
|
||||
assert "/billing" in blob
|
||||
# money-rule: magnitudes-only, never a percentage
|
||||
assert "%" not in blob
|
||||
|
||||
|
||||
def test_money_rule_no_percent():
|
||||
info = _account(
|
||||
paid_service_access=True,
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(
|
||||
subscription_credits_remaining=18.0,
|
||||
purchased_credits_remaining=12.34,
|
||||
total_usable_credits=30.34,
|
||||
),
|
||||
subscription=NousPortalSubscriptionInfo(plan="Pro"),
|
||||
)
|
||||
snap = build_nous_credits_snapshot(info)
|
||||
assert snap is not None
|
||||
for line in snap.details:
|
||||
assert "%" not in line
|
||||
|
||||
|
||||
def test_depleted():
|
||||
info = _account(
|
||||
paid_service_access=False,
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(
|
||||
subscription_credits_remaining=0.0,
|
||||
purchased_credits_remaining=0.0,
|
||||
total_usable_credits=0.0,
|
||||
),
|
||||
subscription=NousPortalSubscriptionInfo(plan="Pro"),
|
||||
)
|
||||
snap = build_nous_credits_snapshot(info)
|
||||
assert snap is not None
|
||||
blob = "\n".join(_all_lines(snap))
|
||||
assert "access depleted" in blob
|
||||
assert "/billing" in blob
|
||||
|
||||
|
||||
def test_purchased_only():
|
||||
info = _account(
|
||||
paid_service_access=True,
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(
|
||||
subscription_credits_remaining=None,
|
||||
purchased_credits_remaining=30.0,
|
||||
total_usable_credits=30.0,
|
||||
),
|
||||
subscription=None,
|
||||
)
|
||||
snap = build_nous_credits_snapshot(info)
|
||||
assert snap is not None
|
||||
blob = "\n".join(_all_lines(snap))
|
||||
assert "Subscription credits" not in blob
|
||||
assert "Top-up credits: $30.00" in blob
|
||||
assert snap.plan is None
|
||||
|
||||
|
||||
def test_logged_out():
|
||||
info = _account(
|
||||
logged_in=False,
|
||||
paid_service_access=True,
|
||||
paid_service_access_info=NousPaidServiceAccessInfo(
|
||||
total_usable_credits=10.0,
|
||||
),
|
||||
)
|
||||
assert build_nous_credits_snapshot(info) is None
|
||||
|
||||
|
||||
def test_none():
|
||||
assert build_nous_credits_snapshot(None) is None
|
||||
|
||||
|
||||
def test_never_raises_empty():
|
||||
info = _account(
|
||||
paid_service_access=True,
|
||||
paid_service_access_info=None,
|
||||
subscription=None,
|
||||
)
|
||||
# No usable numbers and not depleted -> None, without raising.
|
||||
assert build_nous_credits_snapshot(info) is None
|
||||
@@ -0,0 +1,174 @@
|
||||
"""Unit tests for messaging-gateway credit-notice rendering.
|
||||
|
||||
Covers render_notice_line — the pure helper that turns an AgentNotice into the
|
||||
single plaintext line pushed standalone over a messaging platform (no status
|
||||
bar, unlike the TUI). Behavior contracts, not data snapshots.
|
||||
"""
|
||||
from agent.credits_tracker import AgentNotice
|
||||
from gateway.run import render_notice_line
|
||||
|
||||
|
||||
class TestRenderNoticeLine:
|
||||
"""render_notice_line emits the notice text VERBATIM.
|
||||
|
||||
The notice policy already bakes the level glyph (⚠ / • / ✕ / ✓) into the
|
||||
text, and the TUI + CLI REPL render it as-is — so messaging must NOT add a
|
||||
second glyph, which would double it ("⚠ ⚠ Credits 90% used", "⛔ ✕ Credit
|
||||
access paused").
|
||||
"""
|
||||
|
||||
def test_returns_text_verbatim_with_its_baked_glyph(self):
|
||||
assert (
|
||||
render_notice_line(AgentNotice(text="⚠ Credits 90% used · $20.00 cap", level="warn"))
|
||||
== "⚠ Credits 90% used · $20.00 cap"
|
||||
)
|
||||
assert (
|
||||
render_notice_line(AgentNotice(text="• Grant spent · $5.00 top-up left", level="info"))
|
||||
== "• Grant spent · $5.00 top-up left"
|
||||
)
|
||||
assert (
|
||||
render_notice_line(
|
||||
AgentNotice(text="✕ Credit access paused · run /usage for balance", level="error")
|
||||
)
|
||||
== "✕ Credit access paused · run /usage for balance"
|
||||
)
|
||||
|
||||
def test_does_not_prepend_a_second_glyph(self):
|
||||
# Regression: the text already carries its glyph; the level must not add
|
||||
# another (the bug produced "⚠ ⚠ …" / "⛔ ✕ …").
|
||||
line = render_notice_line(AgentNotice(text="⚠ Credits 90% used", level="warn"))
|
||||
assert line == "⚠ Credits 90% used"
|
||||
assert "⚠ ⚠" not in line
|
||||
|
||||
def test_text_is_stripped(self):
|
||||
assert render_notice_line(AgentNotice(text=" ⚠ padded ", level="warn")) == "⚠ padded"
|
||||
|
||||
def test_empty_text_returns_empty_string(self):
|
||||
# Empty/whitespace → "" → the callback suppresses the push. Fail-soft.
|
||||
assert render_notice_line(AgentNotice(text="", level="warn")) == ""
|
||||
assert render_notice_line(AgentNotice(text=" ", level="warn")) == ""
|
||||
|
||||
def test_malformed_notice_does_not_raise(self):
|
||||
# Duck-typed: a stand-in lacking the expected attrs degrades to "".
|
||||
class _Bare:
|
||||
pass
|
||||
|
||||
assert render_notice_line(_Bare()) == ""
|
||||
|
||||
|
||||
def test_real_policy_notices_render_without_doubling():
|
||||
"""End-to-end regression: every notice evaluate_credits_notices emits already
|
||||
carries its glyph, so render_notice_line must return it unchanged (no second
|
||||
glyph prepended) for the messaging push."""
|
||||
from agent.credits_tracker import CreditsState, evaluate_credits_notices
|
||||
|
||||
def _emitted(uf=None, paid=True, purchased=0):
|
||||
latch = {"active": set(), "seen_below_90": True, "usage_band": None}
|
||||
if uf is None:
|
||||
st = CreditsState(
|
||||
subscription_limit_micros=None, subscription_micros=0,
|
||||
denominator_kind="none", paid_access=paid,
|
||||
purchased_micros=purchased, purchased_usd="%.2f" % (purchased / 1e6),
|
||||
)
|
||||
else:
|
||||
lim = 20_000_000
|
||||
st = CreditsState(
|
||||
subscription_limit_micros=lim, subscription_limit_usd="20.00",
|
||||
subscription_micros=int(lim * (1 - uf)), denominator_kind="subscription_cap",
|
||||
paid_access=paid, purchased_micros=purchased,
|
||||
purchased_usd="%.2f" % (purchased / 1e6),
|
||||
)
|
||||
show, _ = evaluate_credits_notices(st, latch)
|
||||
return show
|
||||
|
||||
notices = (
|
||||
_emitted(uf=0.9) # band 90 (warn)
|
||||
+ _emitted(uf=0.5) # band 50 (info)
|
||||
+ _emitted(uf=1.0, purchased=5_000_000) # band 90 + grant_spent
|
||||
+ _emitted(uf=None, paid=False) # depleted
|
||||
)
|
||||
assert notices, "policy produced no notices to check"
|
||||
for n in notices:
|
||||
assert render_notice_line(n) == n.text # verbatim — no prepended glyph
|
||||
|
||||
|
||||
# ── Delivery seam: a rendered notice line goes out via _deliver_platform_notice ──
|
||||
|
||||
import threading
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _make_source(platform_value="telegram", chat_id="555", user_id="u1"):
|
||||
src = MagicMock()
|
||||
plat = MagicMock()
|
||||
plat.value = platform_value
|
||||
src.platform = plat
|
||||
src.chat_id = chat_id
|
||||
src.user_id = user_id
|
||||
return src
|
||||
|
||||
|
||||
def _make_runner_with_adapter(source, adapter):
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = object.__new__(GatewayRunner)
|
||||
runner.adapters = {source.platform: adapter}
|
||||
runner.config = MagicMock()
|
||||
runner.config.get_notice_delivery = MagicMock(return_value="public")
|
||||
runner._thread_metadata_for_source = MagicMock(return_value={"thread": "t"})
|
||||
return runner
|
||||
|
||||
|
||||
class TestDeliverNoticeLine:
|
||||
"""The seam between render_notice_line and the platform adapter.
|
||||
|
||||
Proves a rendered credit-notice line reaches adapter.send (public) /
|
||||
send_private_notice (private) through the shared _deliver_platform_notice
|
||||
rail — the path the gateway notice_callback schedules onto the loop.
|
||||
"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_public_delivery_sends_rendered_line(self):
|
||||
source = _make_source()
|
||||
adapter = MagicMock()
|
||||
adapter.send = AsyncMock(return_value=MagicMock(success=True))
|
||||
runner = _make_runner_with_adapter(source, adapter)
|
||||
|
||||
line = render_notice_line(
|
||||
AgentNotice(text="⚠ Credits 90% used · $20.00 cap", level="warn")
|
||||
)
|
||||
await runner._deliver_platform_notice(source, line)
|
||||
|
||||
adapter.send.assert_awaited_once()
|
||||
args, kwargs = adapter.send.call_args
|
||||
assert args[0] == "555"
|
||||
# Delivered verbatim — the policy's single glyph, not a doubled one.
|
||||
assert args[1] == "⚠ Credits 90% used · $20.00 cap"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_private_delivery_prefers_private_notice(self):
|
||||
source = _make_source()
|
||||
adapter = MagicMock()
|
||||
adapter.send = AsyncMock(return_value=MagicMock(success=True))
|
||||
adapter.send_private_notice = AsyncMock(return_value=MagicMock(success=True))
|
||||
runner = _make_runner_with_adapter(source, adapter)
|
||||
runner.config.get_notice_delivery = MagicMock(return_value="private")
|
||||
|
||||
line = render_notice_line(
|
||||
AgentNotice(text="✓ Credit access restored", level="success")
|
||||
)
|
||||
await runner._deliver_platform_notice(source, line)
|
||||
|
||||
adapter.send_private_notice.assert_awaited_once()
|
||||
adapter.send.assert_not_awaited()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_adapter_is_a_noop(self):
|
||||
source = _make_source()
|
||||
runner = object.__new__(__import__("gateway.run", fromlist=["GatewayRunner"]).GatewayRunner)
|
||||
runner.adapters = {}
|
||||
# Must not raise when the platform has no registered adapter.
|
||||
await runner._deliver_platform_notice(source, "• anything")
|
||||
|
||||
@@ -223,11 +223,14 @@ class TestUsageAccountSection:
|
||||
{"role": "user", "content": "earlier"},
|
||||
]
|
||||
|
||||
calls = {}
|
||||
calls = []
|
||||
|
||||
async def _fake_to_thread(fn, *args, **kwargs):
|
||||
calls["args"] = args
|
||||
calls["kwargs"] = kwargs
|
||||
# /usage dispatches BOTH the account fetch (fetch_account_usage, called
|
||||
# with the provider positionally) and the Nous credits fetch
|
||||
# (nous_credits_lines, markdown-only) through to_thread — record every
|
||||
# call rather than last-wins so we can pick out the account fetch.
|
||||
calls.append({"args": args, "kwargs": kwargs})
|
||||
return fn(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr("gateway.run.asyncio.to_thread", _fake_to_thread)
|
||||
@@ -242,11 +245,14 @@ class TestUsageAccountSection:
|
||||
"Provider: openai-codex (Pro)",
|
||||
],
|
||||
)
|
||||
# The credits block routes through the shared nous_credits_lines() helper;
|
||||
# stub it so this account-section test stays hermetic (no portal/auth lookup).
|
||||
monkeypatch.setattr("agent.account_usage.nous_credits_lines", lambda markdown=False: [])
|
||||
|
||||
event = MagicMock()
|
||||
result = await runner._handle_usage_command(event)
|
||||
|
||||
assert calls["args"] == ("openai-codex",)
|
||||
assert calls["kwargs"]["base_url"] == "https://chatgpt.com/backend-api/codex"
|
||||
account_call = next(c for c in calls if c["args"] == ("openai-codex",))
|
||||
assert account_call["kwargs"]["base_url"] == "https://chatgpt.com/backend-api/codex"
|
||||
assert "📊 **Session Info**" in result
|
||||
assert "📈 **Account limits**" in result
|
||||
|
||||
@@ -0,0 +1,205 @@
|
||||
"""Regression tests for the notice-spine (AgentNotice + emitter callbacks).
|
||||
|
||||
Covers:
|
||||
A. _emit_notice / _emit_notice_clear emitter behaviour (bare AIAgent via
|
||||
object.__new__ — same pattern as test_steer.py and test_file_mutation_verifier.py).
|
||||
B. Constructor / init_agent signature threading.
|
||||
C. TUI _agent_cbs notice binding — mirrors the status_callback tests already
|
||||
in tests/test_tui_gateway_server.py.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from agent.credits_tracker import AgentNotice
|
||||
from run_agent import AIAgent
|
||||
|
||||
|
||||
# ── A. Emitter behaviour ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _bare_agent() -> AIAgent:
|
||||
"""Build an AIAgent without running __init__ (no heavy init required).
|
||||
|
||||
Only the two callback slots used by _emit_notice / _emit_notice_clear are
|
||||
installed — mirrors the pattern in test_steer.py.
|
||||
"""
|
||||
agent = object.__new__(AIAgent)
|
||||
agent.notice_callback = None
|
||||
agent.notice_clear_callback = None
|
||||
return agent
|
||||
|
||||
|
||||
class TestEmitNotice:
|
||||
def test_emit_notice_calls_callback_with_exact_notice(self):
|
||||
agent = _bare_agent()
|
||||
received = []
|
||||
notice = AgentNotice(
|
||||
text="credits 90% used",
|
||||
level="warn",
|
||||
kind="sticky",
|
||||
ttl_ms=None,
|
||||
key="credits.warn90",
|
||||
id="n1",
|
||||
)
|
||||
agent.notice_callback = received.append
|
||||
agent._emit_notice(notice)
|
||||
assert received == [notice]
|
||||
|
||||
def test_emit_notice_clear_calls_callback_with_exact_key(self):
|
||||
agent = _bare_agent()
|
||||
received = []
|
||||
agent.notice_clear_callback = received.append
|
||||
agent._emit_notice_clear("credits.depleted")
|
||||
assert received == ["credits.depleted"]
|
||||
|
||||
def test_emit_notice_swallows_callback_exception(self):
|
||||
agent = _bare_agent()
|
||||
|
||||
def _boom(n):
|
||||
raise RuntimeError("renderer exploded")
|
||||
|
||||
agent.notice_callback = _boom
|
||||
# Must not raise.
|
||||
agent._emit_notice(AgentNotice(text="x"))
|
||||
|
||||
def test_emit_notice_clear_swallows_callback_exception(self):
|
||||
agent = _bare_agent()
|
||||
|
||||
def _boom(key):
|
||||
raise ValueError("clear renderer exploded")
|
||||
|
||||
agent.notice_clear_callback = _boom
|
||||
# Must not raise.
|
||||
agent._emit_notice_clear("some.key")
|
||||
|
||||
def test_emit_notice_no_op_when_callback_is_none(self):
|
||||
agent = _bare_agent()
|
||||
agent.notice_callback = None
|
||||
# Should not raise AttributeError or anything else.
|
||||
agent._emit_notice(AgentNotice(text="x"))
|
||||
|
||||
def test_emit_notice_clear_no_op_when_callback_is_none(self):
|
||||
agent = _bare_agent()
|
||||
agent.notice_clear_callback = None
|
||||
# Should not raise.
|
||||
agent._emit_notice_clear("any.key")
|
||||
|
||||
|
||||
# ── B. Constructor / init_agent signature threading ─────────────────────────
|
||||
|
||||
|
||||
class TestSignatureThreading:
|
||||
def test_agent_init_exposes_notice_callback(self):
|
||||
sig = inspect.signature(AIAgent.__init__)
|
||||
assert "notice_callback" in sig.parameters
|
||||
|
||||
def test_agent_init_exposes_notice_clear_callback(self):
|
||||
sig = inspect.signature(AIAgent.__init__)
|
||||
assert "notice_clear_callback" in sig.parameters
|
||||
|
||||
def test_init_agent_exposes_notice_callback(self):
|
||||
from agent.agent_init import init_agent
|
||||
sig = inspect.signature(init_agent)
|
||||
assert "notice_callback" in sig.parameters
|
||||
|
||||
def test_init_agent_exposes_notice_clear_callback(self):
|
||||
from agent.agent_init import init_agent
|
||||
sig = inspect.signature(init_agent)
|
||||
assert "notice_clear_callback" in sig.parameters
|
||||
|
||||
|
||||
# ── C. TUI _agent_cbs binding ────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAgentCbsNoticeBinding:
|
||||
"""Mirror test_status_callback_emits_kind_and_text from test_tui_gateway_server.py."""
|
||||
|
||||
def test_notice_callback_emits_notification_show(self):
|
||||
from tui_gateway import server
|
||||
|
||||
with patch("tui_gateway.server._emit") as mock_emit:
|
||||
cbs = server._agent_cbs("sid123")
|
||||
notice = AgentNotice(
|
||||
text="credits 90% used",
|
||||
level="warn",
|
||||
kind="sticky",
|
||||
ttl_ms=None,
|
||||
key="credits.warn90",
|
||||
id="n1",
|
||||
)
|
||||
cbs["notice_callback"](notice)
|
||||
|
||||
mock_emit.assert_called_once_with(
|
||||
"notification.show",
|
||||
"sid123",
|
||||
{
|
||||
"text": "credits 90% used",
|
||||
"level": "warn",
|
||||
"kind": "sticky",
|
||||
"ttl_ms": None,
|
||||
"key": "credits.warn90",
|
||||
"id": "n1",
|
||||
},
|
||||
)
|
||||
|
||||
def test_notice_callback_payload_is_full_snake_case_dict(self):
|
||||
"""All six snake_case fields must be present in the payload — no extras,
|
||||
no camelCase variants."""
|
||||
from tui_gateway import server
|
||||
|
||||
captured = []
|
||||
with patch("tui_gateway.server._emit", side_effect=lambda *a: captured.append(a)):
|
||||
cbs = server._agent_cbs("sid123")
|
||||
cbs["notice_callback"](
|
||||
AgentNotice(
|
||||
text="credits 90% used",
|
||||
level="warn",
|
||||
kind="sticky",
|
||||
ttl_ms=None,
|
||||
key="credits.warn90",
|
||||
id="n1",
|
||||
)
|
||||
)
|
||||
|
||||
assert len(captured) == 1
|
||||
_event_type, _sid, payload = captured[0]
|
||||
assert set(payload.keys()) == {"text", "level", "kind", "ttl_ms", "key", "id"}
|
||||
|
||||
def test_notice_clear_callback_emits_notification_clear(self):
|
||||
from tui_gateway import server
|
||||
|
||||
with patch("tui_gateway.server._emit") as mock_emit:
|
||||
cbs = server._agent_cbs("sid123")
|
||||
cbs["notice_clear_callback"]("credits.depleted")
|
||||
|
||||
mock_emit.assert_called_once_with(
|
||||
"notification.clear",
|
||||
"sid123",
|
||||
{"key": "credits.depleted"},
|
||||
)
|
||||
|
||||
def test_notice_callback_event_type_is_notification_show(self):
|
||||
from tui_gateway import server
|
||||
|
||||
captured = []
|
||||
with patch("tui_gateway.server._emit", side_effect=lambda *a: captured.append(a)):
|
||||
cbs = server._agent_cbs("sid123")
|
||||
cbs["notice_callback"](AgentNotice(text="any"))
|
||||
|
||||
assert captured[0][0] == "notification.show"
|
||||
|
||||
def test_notice_clear_callback_event_type_is_notification_clear(self):
|
||||
from tui_gateway import server
|
||||
|
||||
captured = []
|
||||
with patch("tui_gateway.server._emit", side_effect=lambda *a: captured.append(a)):
|
||||
cbs = server._agent_cbs("sid123")
|
||||
cbs["notice_clear_callback"]("some.key")
|
||||
|
||||
assert captured[0][0] == "notification.clear"
|
||||
assert captured[0][1] == "sid123"
|
||||
assert captured[0][2] == {"key": "some.key"}
|
||||
Reference in New Issue
Block a user