Wake quota waits when local capacity changes

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

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,15 +208,15 @@ 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")
                 elif verdict == "telemetry-unavailable":
                     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:

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,20 +2721,64 @@ 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
         fallback. Kiro and OpenRouter agents bypass this entirely (handled by the caller).
 
         `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":