diff --git a/docs/plans/quota-wakeups.md b/docs/plans/quota-wakeups.md new file mode 100644 index 00000000..b4177226 --- /dev/null +++ b/docs/plans/quota-wakeups.md @@ -0,0 +1,132 @@ +# Responsive quota waits + +Implementation: `tauceti_worker/quota_wait.py`, integrated with subscription provider selection in `loop.py` and `quota.py`. The verification and rollout requirements below are release criteria. + +## Problem and intended outcome + +Workers should notice usable capacity within 30 seconds of a file-backed credential change or a new, valid TauCeti usage observation for the current credential. A provider's HTTP retry deadline must continue to govern requests to its usage endpoint independently of how often a sleeping worker checks local state. Platform credential stores and custom quota commands retain their normal lookup schedule; their limitations are specified below. + +The fleet investigation on 2026-10-30 found this sequence: + +- At 06:46 UTC, Claude's usage endpoint returned HTTP 429 with `Retry-After: 3600`. +- At 07:01, worker5 obtained a successful usage response under a changed credential and continued working. Later cached observations showed both Claude windows under pace. +- Workers 4, 6, and 7 retained uninterrupted sleeps until 07:46. They could not notice the changed credential or available shared observation during those waits. +- On waking, those workers cleared the Claude check and encountered exhausted GitHub REST quota instead. That separate limit means early quota recovery does not guarantee an immediate model launch. + +The scheduling defect is the uninterrupted `time.sleep(nap)` in the subscription quota-wait branches of `tauceti_worker/loop.py`. Simply shortening every sleep and re-running `choose_model(..., refresh=True, renew=True)` would increase usage HTTP requests and credential-refresh attempts. Local checks need an explicitly side-effect-free path. + +Evidence: `/home/kim/.local/state/tauceti/fleet-investigations/20261030-quota-sleep/report.json`. + +## Required behaviour + +1. Local observation checks complete at most 30 seconds apart during subscription quota waits, or sooner when an existing eligibility/reset/retry deadline arrives. A tick itself makes no network request, runs no subprocess or external quota command, renews no credential, and spends no model tokens. Include bounded observation time in the cadence rather than adding it after a thirty-second sleep. +2. An unchanged local state keeps the existing provider refresh schedule. Thirty-second checks must not become thirty-second HTTP polling, including for Codex when Claude is the provider whose state changed. +3. A changed credential is evaluated using existing identity and cooldown rules. A token change does not prove an account change and must not clear a known account cooldown. Preserve the existing bounded handling when only a credential-source identity is available. +4. A newly observed cache entry can supply headroom only if the existing validators accept its identity, observation time, TTL, reset clocks, payload, and pacing policy. Time passing cannot turn an old over-pace reading into launch permission. +5. Preserve authentication refusals, known exhaustion until reset, model refusals, bootstrap reservations, forced-provider choices, and the existing `--ignore-quota` semantics. Local observation never broadens permission to launch. +6. Failed or unavailable local observation falls back to the existing scheduled recheck. A missing file, permissions error, corrupt cache, unavailable platform credential store, or failed diagnostic write must not add a new permanent wait. Compare successful observations only: a failed read followed by recovery to the same value is not a credential change. +7. Keep the behaviour automatic. No new service, required configuration, operator-maintained marker, or filesystem notification facility is needed. +8. Read credentials only through the existing supported credential accessors. Existing `/.tauceti-quota/` records and their worker-state fallback are TauCeti-owned and remain in scope. Do not integrate with personal `~/.claude` usage scripts, logs, caches, or coordination mechanisms, and do not add files, locks, or watchers there. +9. Capture absolute provider recheck deadlines once. Local changes may advance those deadlines, but must never postpone the original normal recheck or its existing hourly revisit bound. Time-based release is itself a reason to re-evaluate, even when no file changes. + +## Implementation sequence + +### 1. Establish the current baseline + +Use the current upstream default branch, retaining PR #420 behaviour: plausible HTTP `Retry-After` values are honored in full, and explicit `--ignore-quota` can continue through eligible telemetry outages. Older fifteen-minute caps on explicit server retry headers are not the intended baseline. + +Inspect both subscription quota-wait branches, provider selection, credential lookup, cache validation, cooldown scoping, and launch-stage checks. Preserve their current policies rather than reproducing them in a second implementation. + +### 2. Add an explicit local observation interface + +Add a small interface in `tauceti_worker/quota.py` that reads local provider state without falling back to HTTP or renewing credentials. Do not assume `refresh=False` guarantees this: existing provider methods can fetch on a cache miss. + +Return enough information to identify meaningful changes and reuse the normal branch-specific selection logic: + +- Current credential identity and supported credential-source identity, held privately in memory. +- The relevant usage observation identity and original observation timestamp. +- Applicable cooldown, exhaustion, authentication-refusal, and model-refusal state. +- A provider verdict derived through the existing validators, or an explicit local-unavailable result. + +Inspect Claude's existing primary and worker-state fallback locations without attempting a write or taking the fetch lock. Today the fallback is selected only after a write fails on a particular `Quota` instance; a fresh read-only observer cannot rely on that sticky instance attribute. Both locations supply change hints, not a merged authorization verdict. Keep ordinary lock-driven store selection and refusal semantics unchanged; normal selection after a wake validates the effective store, credential/account scope, and applicable in-process restrictions before deciding what can run. Carry the effective store from an ordinary selection into any cache-only reads that need it; an unknown store is local-unavailable, not permission to fetch or launch. In particular, do not union stale refusals from abandoned locations into a new permanent block: existing auth-refusal records can lack timestamps, so observation time cannot establish their supersession. Test conflicting records as well as a read-only credential mount. + +For isolated homes, extend the supported accessors to inspect the canonical source named by the existing credential-source marker, read-only. The isolated mirror alone cannot reveal an external credential change. A source change is a wake hint; normal provider selection must run the existing `mirror_creds` synchronization before measuring or launching with that credential. Local ticks never rewrite a mirror or credential file. A synchronization failure follows the existing provider policy and must never authorize one identity using another's observation. + +Use a semantic observation signature: identities, validated observation contents and original timestamps, absolute cooldown/refusal deadlines, and relevant bootstrap episode state. Exclude elapsed cache age, countdown values, filesystem modification time alone, and unrelated metadata. Detect atomic replacement with the same timestamp and do not repeatedly wake on identical invalid content. Keep the last successful signature across transient read failures. + +Read bounded files; do not scan directories or histories. On macOS, local ticks check existing cache/cooldown/refusal records for the last established identity without invoking `security`. Keychain credential changes are discovered at the existing normal recheck, with existing timeouts and warning behaviour. This avoids adding repeated Keychain prompts, thirty-second subprocess stalls, or warning spam; the thirty-second file-credential recovery promise does not apply to Keychain-only changes. Pass a captured credential/account snapshot through validators so idle/bootstrap annotation cannot perform hidden credential lookups. + +Wrap the local observation boundary in `except Exception` and treat failures as local-unavailable, retaining the scheduled recheck and applicable known restrictions. Do not catch `BaseException`: stop signals and `KeyboardInterrupt` must propagate. Keep observer reads free of mandatory shared locks. Reconstruct request-scoped provider state or explicitly resolve it each time; do not retain stale `Quota` instance directory/credential memoization across ticks. + +Do not expose tokens, credential contents, or raw usage payloads in status or logs. + +### 3. Replace uninterrupted subscription quota sleeps + +Introduce one wait helper used by the normal paced and explicit `--ignore-quota` branches. Keep the existing provider deadline calculation for when normal provider refresh becomes due. Give the helper an injected wall clock, monotonic clock, and sleeper so tests advance time consistently. + +The helper should: + +1. Capture a successful observer signature immediately before ordinary provider selection and compare it with an observer reading on wait entry. Compare observer output with observer output, rather than comparing the observer's canonical source/both-store view with selection's mirror/effective-store view. This catches changes during selection without treating a persistent source/mirror or primary/fallback disagreement as a new change on every entry. Retain the last successfully handled signature across wake-triggered selection and subsequent blocked iterations; a failed baseline read remains local-unavailable and falls back to scheduled checks. +2. Sleep until the earlier of the next local tick or that deadline, using monotonic time for elapsed waits. Re-evaluate wall-clock reset and retry timestamps through the existing policy when checking state. +3. On a local tick, read the local observation. If it is unchanged and no relevant absolute deadline has arrived, continue waiting without calling normal provider selection. +4. On a meaningful change or deadline, return a wake reason and per-provider refresh permissions to the outer loop. Run existing admission, recovery, branch-specific selection, and launch checks. The helper grants no launch permission. Normal paced selection includes the existing pending-bootstrap path; the explicit `--ignore-quota` branch uses `_ignore_quota_verdict`, including its over-pace and telemetry-outage permissions. Do not substitute `Provider.available` for either branch's existing rules. Allow at most one observation-triggered selection per local tick interval; only an actual scheduled deadline can wake sooner. This bounds persistent mismatches and self-written observations without delaying ordinary due refreshes or imposing a launch gate. +5. If the new state remains blocked, schedule the earliest of the original normal deadline and newly derived deadlines for providers with validated changed local state. Include that identity's scoped cooldown expiry, exhaustion/reset or eligibility boundary, and relevant model-refusal/bootstrap expiry. A cache-only miss, invalid entry, or local-unavailable result preserves that provider's previous absolute deadline and refresh permission; never derive a new `POLL` deadline from a synthetic unavailable sibling verdict. For example, a token change four minutes into a one-hour refusal must be re-evaluated when an applicable fifteen-minute source-scoped pause expires, rather than waiting out the original hour. Repeated peer writes cannot extend the original deadline. +6. If a changed credential has no usable cached observation, permit a refresh only when the existing identity-specific HTTP policy allows it. Preserve a per-provider attempt/deadline state across outer-loop iterations, including failures, so the same missing or malformed sample cannot trigger an attempt on each tick. Coalesce Claude fetches through the existing mechanism. +7. When the original scheduled refresh becomes due, perform the existing normal recheck. Neither failed reads nor successful but still-blocked updates may postpone it. Recompute wall-clock boundaries without repeatedly resetting the monotonic wait budget. Consume a due deadline after its ordinary refresh attempt and derive the next retry from the resulting policy; do not repeatedly wake on a deadline already served. + +Make per-provider refresh control and a genuinely cache-only provider-read mode explicit requirements. A Claude-only wake must not run `choose_model(refresh=True)` for both providers, and `refresh=False` alone is insufficient because cache misses can still fetch. An unchanged sibling with an expired cache remains in its normal scheduled refresh path; an early wake does not create permission to fetch it. Existing callers keep their current default behaviour. + +Claude has shared usage caching; Codex currently has per-worker caches. Preserve that distinction rather than promising shared Codex observations. When several workers detect a change, test both a cacheable Claude response and a winner response with idle or malformed windows that cannot supply a reusable cache. At most one early attempt per worker and observed identity/recovery event is permitted before returning to the existing normal retry schedule; reaching the original scheduled recheck must still work. This bookkeeping is private to a wait episode: it starts at the first blocked selection, survives observation-triggered outer-loop iterations, and ends at a launch or an ordinary refresh at the pinned original deadline. An earlier policy-derived retry consumes its own due deadline and can schedule its next ordinary retry within that original bound. Attempt bookkeeping limits only extra HTTP attempts: it cannot block validated cached recovery, ordinary scheduled refreshes, or otherwise-authorized launches. Add small bounded jitter only to newly advanced network attempts, within the detection budget, to avoid simultaneous bursts. Do not delay cache-backed recovery, introduce a new mandatory lease, or treat jitter as proof of request coalescing. Compare aggregate attempts against the existing scheduling policy for these cases before accepting the implementation. + +Use existing signal handling so stop/restart remains responsive. Avoid filesystem watcher dependencies, background threads, and new persistent leases. The change should add opportunities to resume work without making successful local observation a launch prerequisite. + +### 4. Preserve compatibility and useful status + +- Keep custom `--quota-cmd` execution on its existing schedule; never invoke it on every local tick. With a custom command, sleep to its existing deadline without inspecting unrelated local Claude/Codex records. This change does not invent a cache contract for external commands. +- Keep GitHub quota waits, API funding waits, fleet-capacity admission, no-progress backoff, peer-claim waits, and active model sessions outside this change. +- Keep `next_action_at` as the scheduled normal recheck. The manager uses this field when deciding whether it can reload a worker's read policy; setting it to a local tick and then leaving it stale can defer reloads indefinitely. Report local-check cadence and provider HTTP retry deadlines separately, and let displays derive local-check timing without a durable write on every tick. Test manager reload behaviour while the worker is waiting. +- Record meaningful wait entry, recovery, and normal-refresh events. Avoid a log line or durable status write on every unchanged tick; retain existing supervisor heartbeats. +- Extend the existing usage request journal only if needed to verify behaviour. Logging failure remains non-blocking. + +## Verification + +Use deterministic clocks, simulated local changes, and counted fake HTTP requests. Do not send live requests to provoke a 429. Extend the quota loop, shared usage, cache identity, hard-block, and outage-autonomy tests already in the repository. + +| Scenario | Required result | +| --- | --- | +| One-hour cooldown, unchanged credentials and cache | Local checks continue; zero extra HTTP requests before the original retry deadline; normal recheck still occurs when due. | +| Changed credential plus a valid available shared observation | Worker leaves quota wait within 30 seconds, then passes normal selection and launch checks. | +| Change after blocked selection but before wait entry | Immediate entry observation catches the change; it is not accepted as the unchanged baseline. | +| Persistent source/mirror or primary/fallback disagreement | Observer signatures are compared consistently; at most one observation-triggered selection per tick, with no tight selection/log/renewal loop. | +| Token renewal for an identifiable rate-limited account | The same account's cooldown remains binding; no early endpoint request. | +| Changed credential with only source identity available | Existing bounded source-scoping rules remain intact; the old credential's full cooldown is retained. | +| Source-scoped pause expires after an early token-change observation | An unchanged subsequent signature still wakes at that earlier absolute expiry. | +| Repeated peer writes while still blocked | No update postpones the original normal refresh deadline or existing hourly revisit. | +| Another worker publishes a newer valid Claude observation, while Codex's cache is expired | Waiting peers notice it within 30 seconds; Codex fetch times through its original deadline match the unchanged policy, rather than moving to a new `now + POLL` schedule. | +| Read-only credential mount and fresh `Quota` instance | Cache, cooldown, and refusal records are found in the fallback without writes or locks; conflicting old primary records are reconciled correctly. | +| Isolated home with a changed source token | Source change is noticed without rewriting the mirror on ticks; normal selection synchronizes the mirror before authorization. | +| Old over-pace observation, expired sample, crossed reset, wrong identity, malformed or future-dated observation | No launch permission is manufactured from the entry. | +| Authentication refusal, known exhaustion, model refusal, unopened Claude window | Existing launch restrictions and bootstrap reservation rules remain unchanged. | +| Explicit `--ignore-quota`, including a hard-to-soft transition | Existing permitted outage autonomy and over-pace recovery continue; known hard limits still apply. | +| Missing/unreadable cache, injected unexpected exception, temporary credential lookup failure, unsupported observation source | Existing normal recheck is preserved; no crash, busy loop, or permanent observation-dependent wait. Failure followed by the same successful observation does not count as a change. | +| Same-timestamp replacement, wall-clock adjustment, or restart during a wait | Changes are detected correctly and persisted restrictions remain effective. | +| Several workers recover, with cacheable and uncacheable Claude responses and a Codex credential change | Count aggregate attempts; no repeated early attempts for the same event, no additional polling of unchanged providers, and no bootstrap calls from ticks. | +| macOS locked or slow Keychain | No Keychain subprocess or repeated credential warning on local ticks; the normal bounded lookup schedule is preserved. | +| Manager reload during a long quota wait or just as a tick starts a round | `next_action_at` does not prevent reload indefinitely; existing child/idle checks protect a round that has already started. | +| Forced provider, auto selection, custom quota command, stop signal | Existing provider restrictions and command cadence hold; stop remains prompt. | + +Replace assertions that require a single `sleep(3600)` with elapsed-time and call-count assertions: zero refreshes before the due deadline in an unchanged-state test, then one ordinary recheck. Advance both wall and monotonic clocks with the injected sleeper; a fake sleep that advances neither must not hang the suite. + +During local ticks, make HTTP, credential renewal, credential mirroring, shared-directory creation, fetch-lock acquisition, and subprocess execution fail the test if invoked. Inject ordinary observer exceptions and stop exceptions separately to prove fallback and signal propagation. Include the normal selection and launch path in recovery tests so a cache-only helper cannot pass while the outer loop still refreshes the unchanged sibling. + +Compare request counts with the unchanged implementation for identical unchanged-state and unsuccessful-recovery timelines. The local tick path must issue zero HTTP requests; network requests caused by an actual recovery or launch remain subject to the normal provider policy. Use the existing request journal for rollout evidence. Run the relevant quota integration scripts and the repository's required checks after the focused regressions pass. + +## Review and deployment verification + +Obtain an independent review focused on throughput, autonomous recovery, credential/cooldown identity, refresh frequency, and failure fallback. Require green CI and resolve substantive findings before merging. + +Roll out at an idle boundary without interrupting paid work, preserving worker configuration. Verify installed revision and running process provenance. Observe a natural wait and recovery, or use an isolated fixture with fake credentials and telemetry when a natural event is unavailable. Never rotate personal credentials, edit live cooldowns, or force endpoint rate limits for the demonstration. + +Acceptance requires evidence that supported file/cache recovery is noticed within 30 seconds, applicable HTTP cooldowns are honored, original refresh deadlines cannot be postponed, unchanged-state request frequency has not increased, and failed recovery does not create a new polling loop. Verify all seven workers remain supervised and ordinary work can proceed through the existing launch path. Test container fallback, isolated homes, and macOS behaviour in fixtures as well as the host fleet path. Report unrelated constraints, such as GitHub REST exhaustion, separately so they do not obscure whether quota recovery worked. If live recovery has not occurred, report deployment as installed but behaviour awaiting live confirmation. + +Rollback by restoring the previous reviewed worker release at an idle boundary. No cache migration or cooldown deletion should be needed. Update `docs/quota.md` with the implemented scheduling behaviour and its distinction from HTTP retry timing. diff --git a/docs/quota.md b/docs/quota.md index b5d11365..8c214ab1 100644 --- a/docs/quota.md +++ b/docs/quota.md @@ -90,6 +90,37 @@ effort: a busy lock or unavailable disk never prevents a request or launch. These are TauCeti's records only; they do not count other clients' requests or integrate with external usage scripts or services. +## Local checks during quota waits + +Subscription loops check local credential and TauCeti usage records every 25 +seconds during a quota wait. They can notice a changed file-backed credential or +a new shared Claude observation without waiting out an old, hour-long sleep. +An early network attempt can be staggered by up to five seconds; cached recovery +has no additional delay. These local checks make no HTTP requests, run no +subprocesses, acquire no fetch locks, and do not renew or mirror credentials. +They use existing TauCeti-owned records, including the read-only-mount fallback; +they do not coordinate with personal usage tools. + +An early wake returns through normal provider selection and launch checks. +Each provider retains its own scheduled refresh deadline: a Claude change does +not refresh an unchanged Codex provider, even when Codex's cache has expired. +Repeated local updates cannot postpone the original normal recheck. A newly +applicable earlier cooldown expiry is also checked without needing another file +change. Actual HTTP retries still honor the current identity's persisted +cooldowns. Extra early attempts are bounded within the wait; ordinary due +refreshes and authorized launches remain unaffected. + +Observation errors retain the normal scheduled retry. Isolated workers observe +their canonical credential source and use the existing mirror synchronization +when selection resumes. Keychain-only changes on macOS are discovered at the +normal credential lookup, avoiding additional Keychain prompts or subprocess +polling. Custom `--quota-cmd` commands retain their existing schedule. + +Runtime status keeps `next_action_at` as the normal quota recheck, with the local +check cadence in `quota_local_check_seconds` and that recheck deadline in +`quota_refresh_at`. Local checks do not write status or log a heartbeat on every +tick. GitHub quota waits and no-progress backoff retain their separate schedules. + ## Keeping the Claude token alive: `--auto-refresh` By default the worker never rotates a refresh token. When the Claude access token diff --git a/tauceti_worker/loop.py b/tauceti_worker/loop.py index 41b4ac57..c8cc9eeb 100644 --- a/tauceti_worker/loop.py +++ b/tauceti_worker/loop.py @@ -33,6 +33,7 @@ from .github import github_budget from .github_state import enabled as github_state_enabled from .quota import Provider, Quota, _glyph, _hours, _pace_reason, _unavail_reason, quota_line +from .quota_wait import QuotaWait from .round import run_round_subprocess from .runtime_status import report_runtime, runtime_snapshot @@ -117,6 +118,7 @@ def cmd_loop(args, cfg: Config, *, only: list[str], agent: str, prs: tuple[int, recovery_probe_at = 0.0 collision_streak = 0 previous_sigterm = signal.getsignal(signal.SIGTERM) + quota_wait = QuotaWait(cfg, agent) def terminate(_signum, _frame) -> None: # run_round_subprocess catches KeyboardInterrupt and tears down the round's process @@ -186,7 +188,7 @@ def terminate(_signum, _frame) -> None: # model refusals stop the round and use the existing no-progress backoff. if agent == "auto": raise SystemExit("--ignore-quota --loop needs an explicit --agent (codex/claude)") - _chosen, snap = choose_model(cfg, agent, quota_cmd, refresh=True, renew=True) + _chosen, snap = choose_model(cfg, agent, quota_cmd, refresh=True, renew=True, _wait=quota_wait) prov = snap.get(agent) verdict = _ignore_quota_verdict(_chosen, prov) # An unopened window is the one hard block --ignore-quota may still clear, because the @@ -196,6 +198,8 @@ def terminate(_signum, _frame) -> None: if verdict == "wait": why = prov.error if (prov and prov.error) else (_unavail_reason(prov)[1] if prov else "unavailable") nap = _recheck_in(prov, time.time()) + if quota_wait.deadlines: + nap = max(0, math.ceil(quota_wait.recheck_at() - time.time())) # A loop DOES wait this out, so the wording stays — but a rejected credential is not # something waiting fixes, and an unattended loop can poll on one indefinitely, so # name the command that ends it here too. @@ -204,7 +208,7 @@ def terminate(_signum, _frame) -> None: f"block; sleeping {nap}s{_credential_hint(agent, prov)}" ) report_runtime("waiting-quota", detail=why, next_action_at=time.time() + nap) - time.sleep(nap) + _wait_for_quota(quota_wait, nap, quota_cmd) continue if verdict == "over-pace": log(f"quota: {agent} over-pace; --ignore-quota set — running anyway") @@ -212,7 +216,7 @@ def terminate(_signum, _frame) -> None: log(f"quota: {agent} usage telemetry temporarily unavailable; --ignore-quota set — attempting work") model = agent else: - model, snap = choose_model(cfg, agent, quota_cmd, refresh=True, renew=True) + model, snap = choose_model(cfg, agent, quota_cmd, refresh=True, renew=True, _wait=quota_wait) if model is None and claude_pending_init(snap): model, pending_init = "claude", True if model is None: @@ -225,14 +229,17 @@ def terminate(_signum, _frame) -> None: # well before its reset, so a long wait sleeps to the line, not past it. now = time.time() nap = min((_recheck_in(p, now) for p in snap.values()), default=POLL) + if quota_wait.deadlines: + nap = max(0, math.ceil(quota_wait.recheck_at() - now)) # Neither destination renders Rich markup: log() writes to stderr and a file, and a # runtime-status detail is read back as data. waiting = _wait_quota_line(snap, markup=False) log(f"quota: {waiting} — sleeping {nap}s") report_runtime("waiting-quota", detail=waiting, next_action_at=time.time() + nap) - time.sleep(nap) + _wait_for_quota(quota_wait, nap, quota_cmd) continue + quota_wait.reset() # Mirrored discovery owns a separate read quota. Only a recorded # exhaustion of direct REST should pause a mirrored worker: fresh # dispatch reads and publication still need that credential. Unknown @@ -475,7 +482,13 @@ def _ignore_quota_verdict(chosen: str | None, prov: Provider | None) -> str: def choose_model( - cfg: Config, agent: str, quota_cmd: str | None, *, refresh: bool = False, renew: bool = False + cfg: Config, + agent: str, + quota_cmd: str | None, + *, + refresh: bool = False, + renew: bool = False, + _wait: QuotaWait | None = None, ) -> tuple[str | None, dict]: """Decide which model to run now. With --quota-cmd / TAUCETI_QUOTA_CMD set, consult that external command instead of the built-in pacer (the escape hatch for e.g. a multi-account scheme): run @@ -494,7 +507,21 @@ def choose_model( out = (r.stdout or "").split() model = out[0] if (r.returncode == 0 and out) else None return (model or None), {"quota-cmd": Provider("quota-cmd", bool(model), model)} - return Quota(cfg).choose(None if agent == "auto" else agent, refresh=refresh, renew=renew) + quota = Quota(cfg) + baseline, options = _wait.prepare(quota) if _wait else ({}, {}) + result = quota.choose(None if agent == "auto" else agent, refresh=refresh, renew=renew, **options) + if _wait: + _wait.selected(quota, result[1], baseline, _recheck_in) + return result + + +def _wait_for_quota(wait: QuotaWait, nap: float, quota_cmd: str | None): + if quota_cmd or not wait.deadlines: + time.sleep(nap) + else: + # Keep the manager's actual retry timestamp; local checks do not move it. + report_runtime(next_action_at=wait.recheck_at()) + wait.wait() def _credential_hint(agent: str, prov: Provider | None) -> str: diff --git a/tauceti_worker/quota.py b/tauceti_worker/quota.py index a9811df5..99b540e8 100644 --- a/tauceti_worker/quota.py +++ b/tauceti_worker/quota.py @@ -31,6 +31,11 @@ from .constants import AUTHORING_DEFAULTS, CLAUDE_CMD, POLL from .github import GitHubError + +class LocalUsageUnavailable(Exception): + """An explicitly cache-only selection needs its ordinary scheduled refresh.""" + + CODEX_USAGE_URL = "https://chatgpt.com/backend-api/wham/usage" CLAUDE_USAGE_URL = "https://api.anthropic.com/api/oauth/usage" @@ -970,8 +975,10 @@ def _write_json_atomic(path: Path, data: dict) -> None: def _read_marker(p: Path) -> str | None: try: - return p.read_text().strip() or None - except OSError: + with p.open() as stream: + value = stream.read(4097).strip() + return value if value and len(value) <= 4096 else None + except (OSError, UnicodeError): return None @@ -1233,6 +1240,9 @@ def _raw_path(self, provider: str, fp: str | None = None) -> Path: return directory / f"quota-{provider}{suffix}.json" def _fetch_usage(self, provider: str, url: str, headers: dict, fp: str | None): + if getattr(self, "_cache_only", False): + raise LocalUsageUnavailable + self._usage_attempted = getattr(self, "_usage_attempted", set()) | {provider} from .quota_audit import record self._usage_token_fps = getattr(self, "_usage_token_fps", {}) @@ -1314,6 +1324,24 @@ def launch_credential(self, provider: str) -> str | None: except Exception: return None # unavailable identity must never prevent a launch + def local_credentials(self, provider: str, reader, known: dict): + """Canonical source for local wake hints, without mirroring or Keychain calls. + + The reader supplies bounded JSON reads. Normal launch accessors still + own credential synchronization and verification of the identity used. + """ + if provider == "claude": + source = self._claude_creds_source() + value = ( + known.get(provider) + if sys.platform == "darwin" + else _as_dict(reader(source / ".credentials.json")).get("claudeAiOauth") + ) + else: + source = self._codex_creds_source() + value = reader(source / "auth.json") + return source, value + def _model_refusal_path(self, provider: str, credential: str) -> Path: return self._usage_dir(provider) / f"model-refusal-{provider}-{self._fingerprint(credential)}.json" @@ -1456,7 +1484,7 @@ def _cached_entry(self, provider: str, fp: str | None) -> dict | None: """The whole validated cache entry — {payload, fetched_at, ...} — for a caller that needs to know WHEN the payload was true, not just what it says. See _cached_raw for the payload alone.""" p = self._raw_path(provider, fp) - d = _read_json_file(p) + d = getattr(self, "_cache_reader", _read_json_file)(p) if not isinstance(d, dict): return None # A cache entry is only valid for the SAME account it was fetched under. When the operator rotates @@ -1909,7 +1937,8 @@ def codex(self, *, refresh: bool = False) -> Provider: return self._with_model_refusal("codex", result) def _codex_pass(self, *, refresh: bool = False) -> Provider: - mirror_creds(self.cfg) # re-sync the isolated copy from the operator's fresh file + if not getattr(self, "_cache_only", False): + mirror_creds(self.cfg) # re-sync the isolated copy from the operator's fresh file # The pacer reads auth.json for the token it measures usage with, so it inherits the same # assumption --account does: that auth.json is the credential codex uses. Under a non-file # store it is not, and a leftover file would have us fetch and CACHE one account's usage while @@ -1924,6 +1953,7 @@ def _codex_pass(self, *, refresh: bool = False) -> Provider: error=f"codex credential store is not verifiably the file TauCeti paces against ({store})", ) auth = self._codex_creds() + self._wait_identities = getattr(self, "_wait_identities", {}) | {"codex": auth} if not auth: return Provider("codex", False, None, error="no ~/.codex/auth.json") # Prefer the stable account id so a same-account token refresh keeps the cache; fall back to the @@ -2154,10 +2184,12 @@ def claude(self, *, refresh: bool = False, renew: bool = False) -> Provider: """ # Renew before reading, then mirror, so a rotation reaches an isolated worker's copy in the same # call that performed it. - if renew: + if renew and not getattr(self, "_cache_only", False): self._refresh_claude_credential() - mirror_creds(self.cfg) # re-sync the isolated copy from the operator's fresh file + if not getattr(self, "_cache_only", False): + mirror_creds(self.cfg) # re-sync the isolated copy from the operator's fresh file oauth, _from_keychain = self._claude_creds() + self._wait_identities = getattr(self, "_wait_identities", {}) | {"claude": oauth} if not oauth: err = f"no {claude_dir(self.cfg.home) / '.credentials.json'}" if sys.platform == "darwin": @@ -2230,6 +2262,8 @@ def _claude_usage( ) -> tuple[Provider, list[Reading] | None]: self._usage_token_fps = getattr(self, "_usage_token_fps", {}) self._usage_token_fps["claude"] = self._fingerprint(tok) + if getattr(self, "_cache_only", False): + return self._claude_pass_locked(fp, tok, refresh=False, account=account) # Re-read under the fetch lock: a sibling may have fetched or received a Retry-After while # we waited. The lock covers the network call as well as the cache/cooldown publication. with self._usage_lock(account or fp or "none", fp) as acquired: @@ -2687,7 +2721,17 @@ def _next_eligible(windows: list[Window], now: float | None = None) -> float | N return min(blocked) if blocked else None # --- selection --------------------------------------------------------- - def choose(self, forced: str | None, *, refresh: bool = False, renew: bool = False) -> tuple[str | None, dict]: + def choose( + self, + forced: str | None, + *, + refresh: bool = False, + renew: bool = False, + inspect: set[str] | None = None, + fetch: set[str] | None = None, + previous: dict | None = None, + refresh_providers: set[str] | None = None, + ) -> tuple[str | None, dict]: """Return (agent_to_run_now or None, {codex: Provider, claude: Provider}). forced in {codex, claude}: only that provider counts. None/'auto': codex preferred, opus @@ -2695,12 +2739,46 @@ def choose(self, forced: str | None, *, refresh: bool = False, renew: bool = Fal `renew` is passed through to Quota.claude: a caller that is about to act on the answer may renew an expiring access token, an inspection command may not. + + The loop's wait episode can inspect only changed/due providers, permit + HTTP only for `fetch`, and force refresh only for `refresh_providers`. + Uninspected siblings retain blocked verdicts from `previous`. A cache-only + miss is unknown, never a synthetic telemetry outage that grants autonomy. + Other callers retain the ordinary selection behaviour. """ snap = {} - if forced in (None, "auto", "codex"): - snap["codex"] = self.codex(refresh=refresh) - if forced in (None, "auto", "claude"): - snap["claude"] = self.claude(refresh=refresh, renew=renew) + for provider in ("codex", "claude") if forced in (None, "auto") else (forced,): + if provider not in ("codex", "claude"): + continue + if inspect is not None and provider not in inspect: + # A wait starts with blocked verdicts only. Reusing a restriction + # cannot manufacture headroom when the sibling's cache expires. + old = (previous or {}).get(provider) + snap[provider] = ( + old + if old is not None and not old.available + else Provider(provider, False, None, error="local usage unavailable; awaiting scheduled refresh") + ) + continue + self._cache_only = fetch is not None and provider not in fetch + force = refresh and (refresh_providers is None or provider in refresh_providers) + if provider == "claude": + if not self._cache_only and fetch is not None: + self.__dict__.pop("_claude_usage_directory", None) + elif hasattr(self, "_wait_cache_directory"): + self._claude_usage_directory = self._wait_cache_directory + try: + snap[provider] = ( + self.codex(refresh=force) + if provider == "codex" + else self.claude(refresh=force, renew=renew and not self._cache_only) + ) + except LocalUsageUnavailable: + snap[provider] = Provider( + provider, False, None, error="local usage unavailable; awaiting scheduled refresh" + ) + finally: + self._cache_only = False codex_ok = snap.get("codex") and snap["codex"].available opus_ok = snap.get("claude") and snap["claude"].available if forced == "codex": diff --git a/tauceti_worker/quota_wait.py b/tauceti_worker/quota_wait.py new file mode 100644 index 00000000..48a799d6 --- /dev/null +++ b/tauceti_worker/quota_wait.py @@ -0,0 +1,336 @@ +"""Local quota wake hints. Observation never authorizes work or calls a provider.""" + +from __future__ import annotations + +import hashlib +import json +import math +import random +import sys +import time +from dataclasses import dataclass, field +from pathlib import Path + +from .config import log +from .quota import ( + CLAUDE_BOOTSTRAP_RETRY_S, + CLAUDE_BOOTSTRAP_STALE_MARGIN_S, + CLAUDE_BOOTSTRAP_TIMEOUT_S, + RATE_LIMIT_RECHECK_MAX_S, + Quota, + _cooldown_left, + _finite_num, +) +from .runtime_status import report_runtime + +LOCAL_CHECK_SECONDS = 25 +MAX_FILE_BYTES = 512 * 1024 + + +def _read(path: Path): + try: + with path.open("rb") as stream: + data = stream.read(MAX_FILE_BYTES + 1) + except (FileNotFoundError, NotADirectoryError): + return None + if len(data) > MAX_FILE_BYTES: + raise ValueError("quota observation too large") + return json.loads(data) + + +@dataclass(frozen=True) +class Observation: + signature: str + identity: str + # Compared with the previous observation so expiry is a local wake hint. + # Only the provider's actual verdict can advance its HTTP refresh deadline. + release_at: float | None + headroom: bool = field(default=False, compare=False) + cached: bool = field(default=False, compare=False) + parts: dict = field(default_factory=dict, compare=False, repr=False) + credential: str | None = field(default=None, compare=False, repr=False) + + +def observe( + cfg, provider: str, identities: dict, previous: Observation | None = None, effective_store=None +) -> Observation | None: + """Read fixed, bounded paths only. Exceptions mean a missed hint, never a loop failure.""" + try: + q = Quota(cfg) + parts = {} + + def record(path): + key = str(path) + try: + value = _read(path) + except (OSError, ValueError): + # An unreadable primary store must not hide a readable fallback. + # Nor should a failed read manufacture a change followed by an + # identical recovery. These retained values are hints only. + value = previous.parts.get(key) if previous else None + parts[key] = value + return value + + q._cache_reader = record + # Fully resolved cached windows need no bootstrap annotation for a wake + # hint. Avoid nested credential/ledger reads in this observational path. + q._idle_notes = lambda _readings: ({}, False) + source, credential_data = q.local_credentials(provider, _read, identities) + if provider == "claude": + oauth = credential_data + if not isinstance(oauth, dict) or not oauth.get("accessToken"): + return None + fp = q._fingerprint(oauth["accessToken"]) + account = q._claude_account_key(oauth) + token_fp = fp + directories = [source / ".tauceti-quota" / "usage", q._claude_usage_fallback()] + else: + auth = credential_data + token = ((auth or {}).get("tokens") or {}).get("access_token") + if not token: + return None + token_fp = q._fingerprint(token) + fp = q._codex_account_id(auth) or token_fp + account = fp + directories = [q.cache_dir] + q._usage_token_fps = {provider: token_fp} + records, releases, headroom, has_cache = [], [], False, False + now = time.time() + effective_store = effective_store or q._usage_dir(provider) + for directory in dict.fromkeys(directories): + if provider == "claude": + q._claude_usage_directory = directory + keys = [account, f"forbidden:{token_fp}"] + if provider == "claude" and str(account).startswith("creds:"): + keys.append(f"{account}|credential:{fp}") + cooldown_releases = [] + for key in keys: + cooldown = record(q._cooldown_path(provider, key)) + records.append(cooldown) + # Use the same source/account scoping as actual fetches. Countdown + # values are not part of the signature; expiry is a separate event. + scoped = q._scope_cooldown(cooldown, provider, key, fp if key == account else None) + if directory == effective_store and _cooldown_left(scoped)[0] > 0: + until = scoped.get("until") + cooldown_releases.append(until) + if cooldown_releases: + releases.append(max(cooldown_releases)) + records.extend( + [ + record(q._raw_path(provider, fp)), + record(q._auth_refusal_path(provider, fp)), + ] + ) + # This is only a hint. Selection still resolves the effective store + # and validates its cache before granting any permission. + if provider == "claude": + q._claude_account_key = lambda _oauth=None: account + cached = q._cached_claude(fp) + if cached is not None: + has_cache = True + headroom |= q._from_cached_claude(cached)[0].available + else: + cached = q._cached_codex(fp) + if cached is not None: + has_cache = True + headroom |= q._codex_from_payload(*cached).available + model_releases = [] + for path in q._model_refusal_paths(provider, token_fp): + refusal = record(path) + records.append(refusal) + if isinstance(refusal, dict): + at, until = refusal.get("at"), refusal.get("until") + if _finite_num(at) and _finite_num(until) and at <= now < until <= at + RATE_LIMIT_RECHECK_MAX_S: + model_releases.append(until) + if model_releases: + releases.append(max(model_releases)) + if provider == "claude": + # Bootstrap annotations can change without any new usage sample. + ledger = record(q._bootstrap_paths()[0]) or {} + active = {} + for window in ("session", "weekly"): + key = q._episode_key(window) + reservation = ledger.get(key) + if isinstance(reservation, dict): + active[key] = reservation + at = reservation.get("at") + duration = ( + CLAUDE_BOOTSTRAP_TIMEOUT_S + CLAUDE_BOOTSTRAP_STALE_MARGIN_S + if reservation.get("state") == "in-progress" + else CLAUDE_BOOTSTRAP_RETRY_S + ) + if isinstance(at, (int, float)) and math.isfinite(at) and at <= now < at + duration: + releases.append(at + duration) + records.append(active) + identity = q._fingerprint(fp, token_fp, str(source)) + signature = hashlib.sha256(json.dumps([identity, records], sort_keys=True).encode()).hexdigest() + return Observation(signature, identity, min(releases, default=None), headroom, has_cache, parts, token_fp) + except Exception: + return None + + +class QuotaWait: + """A wait episode retains provider deadlines across opportunistic selections. + + Only extra HTTP attempts are deduplicated. Ordinary deadlines and the outer + loop's admission, provider policy, and launch checks remain authoritative. + """ + + def __init__(self, cfg, agent, *, wall=None, monotonic=None, sleep=None): + self.cfg, self.agent = cfg, agent + self.wall = wall or time.time + self.monotonic = monotonic or time.monotonic + self.sleep = sleep or time.sleep + self.providers = ("codex", "claude") if agent == "auto" else (agent,) + self.identities = {} + self.reset() + + def reset(self): + if getattr(self, "deadlines", None): + report_runtime(quota_local_check_seconds=None, quota_refresh_at=None) + self.deadlines = {} + self.monotonic_deadlines = {} + self.ceilings = {} + self.monotonic_ceilings = {} + self.seen = {} + self.attempted = set() + self.measured = {} + self.selected_credentials = {} + self.previous = {} + self.context = {} + self.pending = None + self.pending_observations = {} + self.last_selection = -math.inf + + def observations(self): + return { + p: observe( + self.cfg, p, self.identities, self.seen.get(p), self.context.get("directory") if p == "claude" else None + ) + for p in self.providers + } + + def prepare(self, quota): + """Called immediately before selection; the returned baseline closes the entry race.""" + baseline = self.observations() + options = {} + if self.pending is not None: + due, changed = self.pending + now, mono = self.wall(), self.monotonic() + due = due | {p for p in self.deadlines if now >= self.deadlines[p] or mono >= self.monotonic_deadlines[p]} + changed = changed | { + p for p, value in baseline.items() if value is not None and p in self.seen and value != self.seen[p] + } + self.pending = due, changed + inspect = due | changed + hints = {p: baseline.get(p) or self.pending_observations.get(p) for p in changed} + fetch = due | { + p for p, value in hints.items() if value is not None and (p, value.identity) not in self.attempted + } + options = dict(inspect=inspect, fetch=fetch, previous=self.previous, refresh_providers=due) + if self.context.get("directory") is not None: + quota._wait_cache_directory = self.context["directory"] + # Cache-only macOS observations must not introduce Keychain polling. + if sys.platform == "darwin" and "claude" not in due and self.identities.get("claude"): + oauth = self.identities["claude"] + quota._claude_creds = lambda: (oauth, True) + quota._refresh_claude_credential = lambda **_kw: False + return baseline, options + + def selected(self, quota, snap, baseline, recheck): + now, mono = self.wall(), self.monotonic() + due = self.pending[0] if self.pending is not None else set(snap) + changed = self.pending[1] if self.pending is not None else set() + inspected = due | changed + self.identities.update(getattr(quota, "_wait_identities", {})) + self.selected_credentials.update(getattr(quota, "_usage_token_fps", {})) + for p, value in getattr(quota, "_wait_identities", {}).items(): + if value is None: + self.selected_credentials[p] = None + if "claude" in inspected: + self.context["directory"] = getattr(quota, "_claude_usage_directory", None) + after = self.observations() + for p, verdict in snap.items(): + if p not in self.deadlines or p in due: + next_at, next_mono = now + recheck(verdict, now), mono + recheck(verdict, now) + if p not in self.ceilings or now >= self.ceilings[p] or mono >= self.monotonic_ceilings[p]: + self.ceilings[p], self.monotonic_ceilings[p] = next_at, next_mono + self.attempted = {key for key in self.attempted if key[0] != p} + self.deadlines[p] = min(next_at, self.ceilings[p]) + self.monotonic_deadlines[p] = min(next_mono, self.monotonic_ceilings[p]) + elif p in changed and not (verdict.error or "").startswith("local usage unavailable"): + self.deadlines[p] = min(self.deadlines[p], now + recheck(verdict, now)) + self.monotonic_deadlines[p] = min(self.monotonic_deadlines[p], mono + recheck(verdict, now)) + if p in inspected: + if baseline.get(p): + self.seen[p] = baseline[p] + elif after.get(p): + # First successful observation establishes a baseline. A + # failed initial read is never itself a recovery event. + self.seen[p] = self._first_observation(p, after[p]) + if p in getattr(quota, "_usage_attempted", set()): + measured = getattr(quota, "_usage_token_fps", {}).get(p) + self.measured[p] = measured + for value in (baseline.get(p), after.get(p), self.pending_observations.get(p)): + if value is not None and measured == value.credential: + self.attempted.add((p, value.identity)) + if after.get(p) and measured == after[p].credential and not after[p].headroom: + # The selection just published its own refusal/blocked + # cache. Do not treat that publication as peer recovery. + self.seen[p] = after[p] + self.previous = snap + self.pending = None + self.pending_observations = {} + self.last_selection = mono + + def wait(self): + original = self.recheck_at() + report_runtime(quota_local_check_seconds=LOCAL_CHECK_SECONDS, quota_refresh_at=original) + # Entry is checked immediately. Thereafter all observation-triggered + # selections are at least one tick apart, even after persistent mismatch. + while True: + now, mono = self.wall(), self.monotonic() + tick_started = mono + due = {p for p in self.deadlines if now >= self.deadlines[p] or mono >= self.monotonic_deadlines[p]} + if due: + self.pending = due, set() + self.pending_observations = {} + return + current = self.observations() + for p, value in current.items(): + if value is not None and p not in self.seen: + self.seen[p] = self._first_observation(p, value) + if self.measured.get(p) == value.credential: + self.attempted.add((p, value.identity)) + changed = {p for p, value in current.items() if value is not None and value != self.seen.get(p)} + if changed and mono >= self.last_selection + LOCAL_CHECK_SECONDS: + # Network-only early wakes get a little staggering; cached recovery + # uses ordinary cache-first selection and is never delayed here. + if any(not current[p].cached and (p, current[p].identity) not in self.attempted for p in changed): + self.sleep(random.uniform(0.01, 5)) + self.pending = set(), changed + self.pending_observations = current + log("quota: local observation changed for " + ", ".join(sorted(changed)) + "; rechecking selection") + return + until = min( + max(0, LOCAL_CHECK_SECONDS - (self.monotonic() - tick_started)), + min(max(0, d - now) for d in self.deadlines.values()), + min(max(0, d - mono) for d in self.monotonic_deadlines.values()), + ) + self.sleep(max(0.01, until)) + + def recheck_at(self): + now, mono = self.wall(), self.monotonic() + remaining = min( + min(max(0, d - now) for d in self.deadlines.values()), + min(max(0, d - mono) for d in self.monotonic_deadlines.values()), + ) + return now + remaining + + def _first_observation(self, provider, value): + if provider in self.selected_credentials and self.selected_credentials[provider] != value.credential: + # Even if the observer missed the initial read, normal selection + # established which credential it measured (or that it was absent). + # A different subsequent credential is a real recovery hint. + return Observation("", "selected credential", None) + return value diff --git a/tests/quota_loop_refresh.py b/tests/quota_loop_refresh.py index ed0d41cb..a5ee829e 100644 --- a/tests/quota_loop_refresh.py +++ b/tests/quota_loop_refresh.py @@ -220,7 +220,7 @@ def _cached(_fp): saved_sleep = tc.loop.time.sleep -def choose(_cfg, agent, _cmd, *, refresh=False, renew=False): +def choose(_cfg, agent, _cmd, *, refresh=False, renew=False, _wait=None): driver_calls.append((agent, refresh, renew)) return "claude", {"claude": tc.Provider("claude", True, "opus")} diff --git a/tests/quota_outage_autonomy.py b/tests/quota_outage_autonomy.py index ced35114..90bd5f04 100644 --- a/tests/quota_outage_autonomy.py +++ b/tests/quota_outage_autonomy.py @@ -356,6 +356,9 @@ def test_cli_preserves_soft_handoff_without_granting_outage_permission(root): with ( patch("tauceti_worker.security_audit.require_admin"), patch("tauceti_worker.github_outbox.drain", return_value=[]), + # This fixture mocks the round; installed agent CLIs are unrelated + # to whether the CLI preserves the operator's pacing permission. + patch.object(cli, "_have", return_value=True), patch.object(cli, "api_mode", return_value=False), patch.object(loop, "choose_model", return_value=(None, {"claude": status})), patch.dict( diff --git a/tests/quota_wakeups.py b/tests/quota_wakeups.py new file mode 100644 index 00000000..6d29a758 --- /dev/null +++ b/tests/quota_wakeups.py @@ -0,0 +1,651 @@ +#!/usr/bin/env python3 +"""Exercise quota wake scheduling with real provider policy and no live requests.""" + +import dataclasses +import json +import os +import sys +import tempfile +from contextlib import ExitStack, contextmanager +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from tauceti_worker import loop, quota, quota_wait + + +class Clock: + def __init__(self): + self.now = 1800000000.0 + self.start = self.now + self.mono = 0.0 + self.events = [] + + def sleep(self, seconds): + assert 0 < seconds <= 30, seconds + self.now += seconds + self.mono += seconds + for at, action in list(self.events): + if self.mono >= at: + self.events.remove((at, action)) + action() + + +def claude_payload(now, used=1): + return { + "five_hour": {"utilization": used, "resets_at": datetime.fromtimestamp(now + 7200, UTC).isoformat()}, + "seven_day": {"utilization": used, "resets_at": datetime.fromtimestamp(now + 3 * 86400, UTC).isoformat()}, + } + + +def codex_payload(): + return { + "rate_limit": { + "primary_window": { + "used_percent": 99, + "limit_window_seconds": 604800, + "reset_after_seconds": 600000, + } + } + } + + +def credentials(home, token="A"): + p = home / ".claude" / ".credentials.json" + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps({"claudeAiOauth": {"accessToken": token, "expiresAt": 9999999999999}})) + + +@contextmanager +def fixture(agent="auto", *, isolated=False, readonly=False): + with tempfile.TemporaryDirectory() as raw, ExitStack() as stack: + root = Path(raw) + home = root / "home" + credentials(home) + (home / ".codex").mkdir() + (home / ".codex/auth.json").write_text(json.dumps({"tokens": {"access_token": "codex", "account_id": "C"}})) + source = root / "source" if isolated else home / ".claude" + if isolated: + source.mkdir() + (source / ".credentials.json").write_text((home / ".claude/.credentials.json").read_text()) + (home / ".claude/.tauceti-creds-source").write_text(str(source)) + if readonly: + (source / ".tauceti-quota").write_text("simulate a read-only credential mount") + cfg = SimpleNamespace( + home=home, state=root / "state/worker", quota_cache=root / "state/worker/cache", wid="test" + ) + clock, calls = Clock(), [] + + def http(url, headers): + provider = "codex" if url == quota.CODEX_USAGE_URL else "claude" + calls.append((provider, clock.mono)) + if provider == "codex": + return 200, codex_payload(), None + if headers["Authorization"] == "Bearer A": + return 429, {}, 3600 + return 200, claude_payload(clock.now), None + + stack.enter_context(patch.dict(os.environ, {"TAUCETI_GITHUB_STATE_URL": ""})) + for key in ("CLAUDE_CONFIG_DIR", "CODEX_HOME", "TAUCETI_AUTO_REFRESH"): + os.environ.pop(key, None) + stack.enter_context(patch.object(quota.time, "time", side_effect=lambda: clock.now)) + stack.enter_context(patch.object(quota.time, "monotonic", side_effect=lambda: clock.mono)) + stack.enter_context(patch.object(quota, "_http_get_json", side_effect=http)) + stack.enter_context(patch.object(quota_wait.random, "uniform", return_value=1.0)) + stack.enter_context(patch.object(loop, "report_runtime")) + stack.enter_context(patch.object(quota_wait, "report_runtime")) + quota._COOLDOWNS.clear() + wait = quota_wait.QuotaWait(cfg, agent, wall=lambda: clock.now, monotonic=lambda: clock.mono, sleep=clock.sleep) + yield SimpleNamespace(cfg=cfg, clock=clock, calls=calls, wait=wait, source=source, home=home) + + +def select(f, agent="auto"): + return loop.choose_model(f.cfg, agent, None, refresh=True, renew=True, _wait=f.wait) + + +def publish(f, token="B", used=1): + (f.source / ".credentials.json").write_text(json.dumps({"claudeAiOauth": {"accessToken": token}})) + q = quota.Quota(f.cfg) + if f.source != f.home / ".claude": + # Cache publication belongs to another worker using the canonical source. + q._claude_usage_directory = f.source / ".tauceti-quota/usage" + if (f.source / ".tauceti-quota").is_file(): + q._claude_usage_directory = q._claude_usage_fallback() + q._store_raw("claude", claude_payload(f.clock.now, used), q._fingerprint(token), f.clock.now + 7200) + + +def test_original_deadline_and_no_side_effects(): + with fixture() as f: + assert select(f)[0] is None + original = dict(f.wait.deadlines) + with ExitStack() as s: + traps = [] + for target in ("_http_get_json", "mirror_creds"): + traps.append(s.enter_context(patch.object(quota, target, side_effect=AssertionError(target)))) + traps.append(s.enter_context(patch.object(quota.Quota, "_usage_lock", side_effect=AssertionError("lock")))) + traps.append( + s.enter_context(patch.object(quota.subprocess, "run", side_effect=AssertionError("subprocess"))) + ) + traps.append(s.enter_context(patch.object(Path, "mkdir", side_effect=AssertionError("mkdir")))) + f.wait.wait() + for trap in traps: + trap.assert_not_called() + assert f.clock.mono == 3600 + assert f.wait.deadlines == original + assert f.calls == [("codex", 0), ("claude", 0)] + select(f) + assert [p for p, at in f.calls if at == 3600] == ["codex", "claude"] + + +def test_shared_recovery(*, isolated=False, readonly=False): + with fixture(isolated=isolated, readonly=readonly) as f: + assert select(f)[0] is None + deadline = f.wait.deadlines["codex"] + f.clock.events.append((100, lambda: publish(f))) + f.wait.wait() + assert 100 <= f.clock.mono <= 130 + assert select(f)[0] == "claude" + assert f.wait.deadlines["codex"] == deadline + assert f.calls == [("codex", 0), ("claude", 0)], f.calls + if isolated: + assert json.loads((f.home / ".claude/.credentials.json").read_text())["claudeAiOauth"]["accessToken"] == "B" + + +def test_source_cooldown_expiry(): + with fixture() as f: + assert select(f)[0] is None + f.clock.events.append((100, lambda: credentials(f.home, "B"))) + f.wait.wait() + assert select(f)[0] is None + assert f.wait.deadlines["claude"] == f.clock.start + 900 + while f.clock.mono < 900: + f.wait.wait() + model, _ = select(f) + assert model == "claude" + assert f.clock.mono == 900 + assert f.calls == [("codex", 0), ("claude", 0), ("claude", 900)] + + +def test_sibling_miss_and_repeated_updates(): + with fixture() as f: + select(f) + deadline = f.wait.deadlines["codex"] + for i in range(1, 25): + f.clock.events.append((i * 100, lambda: publish(f, "A", used=99))) + while f.clock.mono < 3600: + f.wait.wait() + select(f) + assert f.wait.deadlines["codex"] == deadline or f.clock.mono == 3600 + assert [at for p, at in f.calls if p == "codex"] == [0, 3600], f.calls + + +def test_entry_race_and_persistent_mirror_failure(): + with fixture(isolated=True) as f: + select(f) + publish(f) + with patch.object(quota, "mirror_creds", return_value=None): + for _ in range(4): + previous = f.clock.mono + f.wait.wait() + assert f.clock.mono >= previous + 25 + select(f) + assert [p for p, _ in f.calls].count("claude") <= 2 + + +def test_failed_observation_and_signals(): + with fixture() as f: + select(f) + with patch.object(quota_wait, "observe", return_value=None): + f.wait.wait() + assert f.clock.mono == 3600 + f.wait.reset() + select(f) + with patch.object(quota_wait, "_read", side_effect=RuntimeError("bad peer shape")): + f.wait.wait() + assert f.clock.mono == 7200 + with patch.object(quota_wait, "_read", side_effect=KeyboardInterrupt): + try: + f.wait.observations() + raise AssertionError("stop swallowed") + except KeyboardInterrupt: + pass + + +def test_same_account_and_same_timestamp(): + with fixture() as f: + # A stable account identity cannot escape its full Retry-After by token rotation. + path = f.source / ".credentials.json" + path.write_text(json.dumps({"claudeAiOauth": {"accessToken": "A", "accountUuid": "same"}})) + select(f) + before = quota_wait.observe(f.cfg, "claude", {}) + stamp = path.stat().st_mtime_ns + path.write_text(json.dumps({"claudeAiOauth": {"accessToken": "B", "accountUuid": "same"}})) + os.utime(path, ns=(stamp, stamp)) + after = quota_wait.observe(f.cfg, "claude", {}) + assert before.signature != after.signature + f.wait.wait() + assert select(f)[0] is None + assert f.wait.deadlines["claude"] == f.clock.start + 3600 + assert len(f.calls) == 2 + + +def test_ignore_quota_and_cache_only(): + with fixture(agent="claude") as f: + select(f, "claude") + f.clock.events.append((100, lambda: publish(f, used=99))) + f.wait.wait() + model, snap = select(f, "claude") + assert model is None + assert loop._ignore_quota_verdict(model, snap["claude"]) == "over-pace" + q = quota.Quota(f.cfg) + with patch.object(quota, "_http_get_json", side_effect=AssertionError("network")): + model, snap = q.choose("codex", inspect={"codex"}, fetch=set()) + assert model is None + assert not snap["codex"].telemetry_unavailable + + +def test_keychain_and_custom_command(): + with fixture(agent="claude") as f: + select(f, "claude") + with ( + patch.object(quota_wait.sys, "platform", "darwin"), + patch.object(quota.subprocess, "run", side_effect=AssertionError("Keychain")), + ): + assert f.wait.observations()["claude"] is not None + with ( + patch.object(quota_wait, "observe", side_effect=AssertionError("custom observation")), + patch.object(loop.time, "sleep") as sleep, + ): + loop._wait_for_quota(f.wait, 3600, "custom-quota") + sleep.assert_called_once_with(3600) + + +def test_outer_loop_reaches_launch(*, ignore=False, stop_wait=False): + from tauceti_worker import config, repair_recovery, verification_jobs + + with fixture(agent="claude" if ignore else "auto") as f, ExitStack() as stack: + if ignore: + + def http(*_args): + return 200, claude_payload(f.clock.now, used=100), None + + stack.enter_context(patch.object(quota, "_http_get_json", side_effect=http)) + f.clock.events.append((100, lambda: publish(f, used=99 if ignore else 1))) + for module, name, result in ( + (config, "rotate_console_log", None), + (verification_jobs, "reap_continuations", None), + (verification_jobs, "resume_pending", None), + (repair_recovery, "due", False), + (loop, "fleet_admission_reason", None), + (loop, "github_budget", {}), + ): + stack.enter_context(patch.object(module, name, return_value=result)) + stack.enter_context( + patch.object(loop.time, "sleep", side_effect=loop._LoopTerminated if stop_wait else f.clock.sleep) + ) + launched = [] + + def launch(tail): + launched.append(tail) + raise KeyboardInterrupt + + stack.enter_context(patch.object(loop, "run_round_subprocess", side_effect=launch)) + result = loop.cmd_loop( + SimpleNamespace(ignore_quota=ignore, bubble=False, quota_cmd=None), + f.cfg, + only=["review"], + agent="claude" if ignore else "auto", + ) + if stop_wait: + assert result == 143 and not launched + return + assert result == 130 + assert 100 <= f.clock.mono <= 130, f.clock.mono + assert len(launched) == 1 + assert launched[0][launched[0].index("--agent") + 1] == "claude" + assert ("--ignore-quota" in launched[0]) == ignore + if not ignore: + assert f.calls == [("codex", 0), ("claude", 0)] + + +def test_wall_clock_and_original_ceiling(): + with fixture() as f: + select(f) + original = f.wait.deadlines["claude"] + f.clock.events.append((100, lambda: credentials(f.home, "B"))) + f.wait.wait() + select(f) + f.wait.wait() + # A failure at the advanced deadline must not discard the original hourly visit. + with patch.object(quota, "_http_get_json", return_value=(429, {}, 7200)): + select(f) + assert f.wait.deadlines["claude"] <= original + f.clock.now -= 5000 + while f.clock.mono < 3600: + f.wait.wait() + if f.clock.mono < 3600: + select(f) + assert f.clock.mono == 3600 + + +def test_no_uncacheable_response_storm(): + with fixture() as f: + select(f) + credentials(f.home, "B") + f.wait.wait() + select(f) # the source-scoped pause still applies + f.wait.wait() + with patch.object(quota, "_http_get_json", return_value=(200, {}, None)) as http: + select(f) + before = f.clock.mono + f.wait.wait() + assert f.clock.mono >= before + 300 + assert http.call_count == 1 + + +def test_sibling_due_and_admission_interleavings(): + with fixture() as f: + select(f) + f.clock.sleep(10) + publish(f) + f.wait.deadlines["codex"] = f.clock.start + 20 + f.wait.monotonic_deadlines["codex"] = 20 + f.wait.wait() + assert f.clock.mono == 20 + assert select(f)[0] == "claude" + with fixture() as f: + select(f) + f.clock.events.append((100, lambda: publish(f))) + f.wait.wait() + # Simulate another credential change while the outer admission path waits. + path = f.home / ".codex/auth.json" + path.write_text(json.dumps({"tokens": {"access_token": "new", "account_id": "D"}})) + assert select(f)[0] == "claude" + assert f.calls[-1] == ("codex", 100) + + +def test_first_failed_observation_and_macos_selection(): + with fixture() as f: + with patch.object(quota_wait, "observe", return_value=None): + select(f) + f.wait.wait() + assert f.clock.mono == 3600 + assert len(f.calls) == 2 + with fixture(agent="claude") as f: + select(f, "claude") + known = dict(f.wait.identities) + f.wait.reset() + assert f.wait.identities == known + f.wait = quota_wait.QuotaWait( + f.cfg, "claude", wall=lambda: f.clock.now, monotonic=lambda: f.clock.mono, sleep=f.clock.sleep + ) + with ( + patch.object(quota_wait.sys, "platform", "darwin"), + patch.object(quota.Quota, "_claude_creds", return_value=({"accessToken": "A"}, True)), + ): + select(f, "claude") + f.clock.events.append((100, lambda: publish(f, "A"))) + with ( + patch.object(quota_wait.sys, "platform", "darwin"), + patch.object( + quota, "_claude_keychain_creds", side_effect=AssertionError("early Keychain call") + ) as keychain, + ): + f.wait.wait() + assert select(f, "claude")[0] == "claude" + keychain.assert_not_called() + assert len(f.calls) == 1 + + +def test_unreadable_primary_and_observation_recovery(): + with fixture(readonly=True) as f: + select(f) + original_read = quota_wait._read + primary = f.source / ".tauceti-quota/usage" + + def read(path): + if str(path).startswith(str(primary)): + raise PermissionError("unreadable primary") + return original_read(path) + + with patch.object(quota_wait, "_read", side_effect=read): + f.clock.events.append((100, lambda: publish(f))) + f.wait.wait() + assert select(f)[0] == "claude" + assert len(f.calls) == 2 + with fixture() as f: + select(f) + baseline = f.wait.observations() + for p in baseline: + f.wait.seen[p] = baseline[p] + original_read = quota_wait._read + + def temporarily_unreadable(path): + if path.name.startswith("quota-"): + raise PermissionError("peer replacing cache") + return original_read(path) + + with patch.object(quota_wait, "_read", side_effect=temporarily_unreadable): + assert f.wait.observations() == baseline + assert f.wait.observations() == baseline + + +def test_manager_status_and_signal_in_sleep(): + from tauceti_worker import worker_manager as wm + + with fixture() as f: + select(f) + f.clock.events.append((100, lambda: publish(f))) + with patch.object(loop, "report_runtime") as status, patch.object(quota_wait, "report_runtime") as local: + loop._wait_for_quota(f.wait, 3600, None) + assert status.call_args.kwargs["next_action_at"] == f.clock.start + 3600 + assert local.call_args.kwargs["quota_local_check_seconds"] == 25 + spec = wm.WorkerSpec("test") + desired = dataclasses.replace(spec, max_open_prs=33) + live = {"state": "waiting-quota", "applied_spec": spec.as_dict(), "next_action_at": f.wait.recheck_at()} + assert not wm._defer_read_policy_reload(desired, live) + live["state"] = "surveying" + assert wm._defer_read_policy_reload(desired, live) + with patch.object(quota_wait, "report_runtime") as local: + f.wait.reset() + assert local.call_args.kwargs == {"quota_local_check_seconds": None, "quota_refresh_at": None} + with fixture() as f: + select(f) + f.wait.sleep = lambda _seconds: (_ for _ in ()).throw(loop._LoopTerminated()) + try: + f.wait.wait() + raise AssertionError("SIGTERM swallowed") + except loop._LoopTerminated: + pass + + +def test_restart_and_restrictions_in_cache_only(): + with fixture(agent="claude") as f: + select(f, "claude") + quota._COOLDOWNS.clear() + f.wait = quota_wait.QuotaWait( + f.cfg, "claude", wall=lambda: f.clock.now, monotonic=lambda: f.clock.mono, sleep=f.clock.sleep + ) + assert select(f, "claude")[0] is None + assert f.wait.deadlines["claude"] == f.clock.start + 3600 + assert len(f.calls) == 1 + publish(f, "A") + q = quota.Quota(f.cfg) + credential = q.launch_credential("claude") + q.note_model_refusal("claude", credential) + model, snap = q.choose("claude", inspect={"claude"}, fetch=set()) + assert model is None and snap["claude"].model_refusal + q._raw_path("claude", credential).unlink() + refusal = q._auth_refusal_path("claude", credential) + refusal.write_text(json.dumps({"code": 401})) + model, snap = q.choose("claude", inspect={"claude"}, fetch=set()) + assert model is None and "credential rejected" in snap["claude"].error + assert not snap["claude"].telemetry_unavailable + + +def test_concurrent_recovery_attempt_counts(*, cacheable=True): + with fixture() as f: + peers = [] + for i in range(4): + cfg = SimpleNamespace(**vars(f.cfg)) + cfg.wid = f"worker{i}" + cfg.quota_cache = f.cfg.state.parent / cfg.wid / "cache" + cfg.state = cfg.quota_cache.parent + wait = quota_wait.QuotaWait( + cfg, "auto", wall=lambda: f.clock.now, monotonic=lambda: f.clock.mono, sleep=f.clock.sleep + ) + assert loop.choose_model(cfg, "auto", None, refresh=True, renew=True, _wait=wait)[0] is None + peers.append((cfg, wait)) + assert [p for p, _ in f.calls].count("claude") == 1 + f.clock.sleep(25) + f.clock.sleep(25) + if cacheable: + publish(f) + for cfg, wait in peers: + wait.wait() + assert loop.choose_model(cfg, "auto", None, refresh=True, renew=True, _wait=wait)[0] == "claude" + assert [p for p, _ in f.calls].count("claude") == 1 + assert [p for p, _ in f.calls].count("codex") == 4 + else: + (f.source / ".credentials.json").write_text( + json.dumps({"claudeAiOauth": {"accessToken": "B", "accountUuid": "new"}}) + ) + with patch.object(quota, "_http_get_json", return_value=(200, {}, None)) as http: + for cfg, wait in peers: + wait.wait() + loop.choose_model(cfg, "auto", None, refresh=True, renew=True, _wait=wait) + assert http.call_count == 4 + assert all(wait.deadlines["claude"] >= f.clock.start + 350 for _, wait in peers) + + +def test_conflicting_stores_keep_effective_authority(): + with fixture() as f: + primary = f.source / ".tauceti-quota/usage" + primary.mkdir(parents=True) + original_open = quota.os.open + + def open_file(path, *args, **kwargs): + if str(path).startswith(str(primary)) and Path(path).name.startswith("fetch-"): + raise PermissionError("read-only primary") + return original_open(path, *args, **kwargs) + + with patch.object(quota.os, "open", side_effect=open_file): + select(f) + assert f.wait.context["directory"] == quota.Quota(f.cfg)._claude_usage_fallback() + q = quota.Quota(f.cfg) + fp = q._fingerprint("A") + q._auth_refusal_path("claude", fp).write_text(json.dumps({"code": 401})) + q._claude_usage_directory = q._claude_usage_fallback() + q._store_raw("claude", claude_payload(f.clock.now), fp, f.clock.now + 7200) + f.wait.wait() + assert select(f)[0] == "claude" + assert len(f.calls) == 2 + + +def test_failed_baseline_still_notices_real_credential_race(): + with fixture(agent="claude") as f: + original_observe = quota_wait.observe + calls = [0] + + def observe(*args, **kwargs): + calls[0] += 1 + return None if calls[0] == 1 else original_observe(*args, **kwargs) + + def http(*_args): + f.calls.append(("claude", f.clock.mono)) + publish(f) + return 429, {}, 3600 + + with ( + patch.object(quota_wait, "observe", side_effect=observe), + patch.object(quota, "_http_get_json", side_effect=http), + ): + assert select(f, "claude")[0] is None + f.wait.wait() + assert f.clock.mono <= 30 + assert select(f, "claude")[0] == "claude" + assert len(f.calls) == 1 + + +def test_empty_bootstrap_period_is_not_a_change(): + with fixture(agent="claude") as f: + f.clock.now += 17900 + select(f, "claude") + f.wait.wait() + assert f.clock.mono == 3600 + assert len(f.calls) == 1 + + +def test_pending_hint_survives_transient_prepare_failure(): + with fixture(agent="claude") as f: + select(f, "claude") + (f.source / ".credentials.json").write_text( + json.dumps({"claudeAiOauth": {"accessToken": "B", "accountUuid": "new"}}) + ) + f.wait.wait() + original_observe, count = quota_wait.observe, [0] + + def observe(*args, **kwargs): + count[0] += 1 + return None if count[0] == 1 else original_observe(*args, **kwargs) + + with patch.object(quota_wait, "observe", side_effect=observe): + assert select(f, "claude")[0] == "claude" + assert len(f.calls) == 2 + assert f.clock.mono <= 30 + + +def test_pending_attempt_survives_two_failed_reads(): + with fixture(agent="claude") as f: + select(f, "claude") + (f.source / ".credentials.json").write_text( + json.dumps({"claudeAiOauth": {"accessToken": "B", "accountUuid": "new"}}) + ) + f.wait.wait() + with ( + patch.object(quota_wait, "observe", return_value=None), + patch.object(quota, "_http_get_json", return_value=(200, {}, None)) as http, + ): + assert select(f, "claude")[0] is None + assert http.call_count == 1 + f.wait.wait() + with patch.object(quota, "_http_get_json", side_effect=AssertionError("duplicate early request")) as http: + assert select(f, "claude")[0] is None + http.assert_not_called() + + +if __name__ == "__main__": + for test in ( + test_original_deadline_and_no_side_effects, + test_shared_recovery, + lambda: test_shared_recovery(isolated=True), + lambda: test_shared_recovery(readonly=True), + test_source_cooldown_expiry, + test_sibling_miss_and_repeated_updates, + test_entry_race_and_persistent_mirror_failure, + test_failed_observation_and_signals, + test_same_account_and_same_timestamp, + test_ignore_quota_and_cache_only, + test_keychain_and_custom_command, + test_outer_loop_reaches_launch, + lambda: test_outer_loop_reaches_launch(ignore=True), + test_wall_clock_and_original_ceiling, + test_no_uncacheable_response_storm, + test_sibling_due_and_admission_interleavings, + test_first_failed_observation_and_macos_selection, + test_unreadable_primary_and_observation_recovery, + test_manager_status_and_signal_in_sleep, + test_restart_and_restrictions_in_cache_only, + test_concurrent_recovery_attempt_counts, + lambda: test_concurrent_recovery_attempt_counts(cacheable=False), + test_conflicting_stores_keep_effective_authority, + test_failed_baseline_still_notices_real_credential_race, + test_empty_bootstrap_period_is_not_a_change, + test_pending_hint_survives_transient_prepare_failure, + lambda: test_outer_loop_reaches_launch(stop_wait=True), + test_pending_attempt_survives_two_failed_reads, + ): + test() + print("PASS", test.__name__) diff --git a/tests/run-all b/tests/run-all index f232f161..37324b1e 100755 --- a/tests/run-all +++ b/tests/run-all @@ -117,6 +117,7 @@ RUN_STDLIB=( quota_auto_refresh.py quota_cache_fp.py quota_loop_refresh.py + quota_wakeups.py quota_shared_usage.py rebase_handoff.py scoped_repair_route.py