From e56ec8c9f7a275e9b6892627e941498155d44bf7 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Fri, 18 Sep 2026 19:16:27 -0700 Subject: [PATCH] fix(chronos): a 403 invalid_client from NAS hands cron fires to the built-in ticker (#97494) NAS maps the agent-cron bearer to a provisioned instance via an `agent:*` client or the `hermes-cli-vps` bootstrap session (hermes-portal `server/agent-cron/instance-auth.ts`). A container whose auth.json holds a plain `hermes-cli` user login is refused with 403 invalid_client on every arm, re-arm and list, for the life of that credential - and the re-login users try first (`hermes auth logout nous` + device code) replaces the bootstrap session, making it permanent. Chronos previously logged one bare warning per job and left the jobs with no trigger at all: they only ran through the misfire sweep, minutes late. NasCronClientError now carries the HTTP status and the OAuth `error` code. On an identity rejection the provider logs ONE warning that names the real remedy (restore the hosted credential from the Nous Portal; re-login cannot fix it), stops calling NAS, and starts the built-in ticker with the gateway's own adapters/loop so scheduled jobs keep firing on time. Transient 5xx/transport failures keep retrying on the next reconcile. Supersedes #97566 (@wesleysimplicio), whose runtime-credential swap resolves to the same bearer on main (`agent_key` is the access token) and so could not change the 403. --- plugins/cron_providers/chronos/__init__.py | 51 +++++++++++++- plugins/cron_providers/chronos/_nas_client.py | 24 ++++++- tests/cron/test_ticker_startup_survival.py | 14 ++++ tests/plugins/test_chronos_cron.py | 66 +++++++++++++++++++ .../chronos-managed-cron-contract.md | 14 ++++ 5 files changed, 165 insertions(+), 4 deletions(-) diff --git a/plugins/cron_providers/chronos/__init__.py b/plugins/cron_providers/chronos/__init__.py index 0af25f24bc..9a846aa73d 100644 --- a/plugins/cron_providers/chronos/__init__.py +++ b/plugins/cron_providers/chronos/__init__.py @@ -15,6 +15,8 @@ from typing import Any, Dict from cron.scheduler_provider import CronScheduler +from ._nas_client import NasCronClientError + logger = logging.getLogger("cron.chronos") @@ -35,6 +37,12 @@ class ChronosCronScheduler(CronScheduler): self._armed: Dict[str, str] = {} self._lock = threading.Lock() self._client = None # lazily constructed (no network in is_available) + # Set when NAS answered 403 invalid_client: the Nous token in auth.json is not this + # instance's provisioned identity, so every arm would fail the same way for the life of + # the process. Once set, NAS is left alone and the built-in ticker fires jobs (#97494). + self._identity_rejected = False + self._stop_event = None + self._ticker_kwargs: Dict[str, Any] = {} @property def name(self) -> str: @@ -66,6 +74,10 @@ class ChronosCronScheduler(CronScheduler): def start(self, stop_event, *, adapters=None, loop=None, interval=60): """Arm all enabled jobs via NAS, then RETURN — no loop, no periodic wake (scale-to-zero).""" + # Kept so a later identity rejection (boot or mid-life re-arm) can hand this process's + # fires to the built-in ticker with the gateway's own adapters/loop. + self._ticker_kwargs = {"adapters": adapters, "loop": loop, "interval": interval} + self._stop_event = stop_event # A new lifecycle can't prove what an interrupted process did: classify unknown, never requeue. self.recover_interrupted() self._reconcile_logged(logger.warning, "start()") @@ -74,15 +86,23 @@ class ChronosCronScheduler(CronScheduler): pass def on_jobs_changed(self) -> None: - self._reconcile_logged(logger.debug, "on_jobs_changed") + if not self._identity_rejected: + self._reconcile_logged(logger.debug, "on_jobs_changed") def register_job(self, job: Dict[str, Any]) -> None: """Arm the first one-shot for a new job; may raise so creation can report it.""" - self._arm_one_shot(job) + try: + self._arm_one_shot(job) + except NasCronClientError as e: + if not e.identity_rejected: + raise + self._note_identity_rejected() # the job is stored; the ticker fires it def _arm_one_shot(self, job: Dict[str, Any]) -> None: """Arm one one-shot at next_run_at (agent computes the time; NAS executes). dedup_key=(job_id, fire_at) makes re-arming the same fire a no-op.""" + if self._identity_rejected: + return # the built-in ticker owns this process's fires; NAS would 403 again job_id = job["id"] fire_at = job.get("next_run_at") if not fire_at: @@ -93,11 +113,38 @@ class ChronosCronScheduler(CronScheduler): with self._lock: self._armed[job_id] = fire_at + def _note_identity_rejected(self) -> None: + """403 invalid_client is deterministic: NAS maps the bearer to a provisioned instance via an + ``agent:*`` client or the hosted bootstrap session, and a plain ``hermes auth`` login is + neither — re-logging in cannot fix it, which is what users try first (#97494). Without + NAS the jobs have no trigger at all (the misfire sweep runs them ``misfire_grace_minutes`` + late), so the built-in ticker takes over this process's fires.""" + with self._lock: + if self._identity_rejected: + return + self._identity_rejected = True + logger.warning( + "Chronos: NAS rejected this agent's Nous credential for agent-cron (403 invalid_client). " + "The Nous token in auth.json is not this instance's provisioned identity (an agent:* client " + "or the hosted bootstrap session), so no job can be armed. A normal `hermes auth` re-login " + "cannot fix this; the hosted credential has to be restored from the Nous Portal. Falling back " + "to the built-in cron ticker for this process so scheduled jobs keep firing on time.") + if self._stop_event is None: + return # start() never ran (e.g. a CLI `hermes cron add`); nothing to tick here + from agent.memory_provider import spawn_context_thread + from cron.scheduler_provider import InProcessCronScheduler + spawn_context_thread( + InProcessCronScheduler().start, name="cron-scheduler-chronos-fallback", + args=(self._stop_event,), kwargs=self._ticker_kwargs).start() + def _arm_logged(self, job: Dict[str, Any], what: str) -> None: """Best-effort arm: log a warning instead of raising (reconcile/fire must not die).""" try: self._arm_one_shot(job) except Exception as e: + if isinstance(e, NasCronClientError) and e.identity_rejected: + self._note_identity_rejected() + return logger.warning("Chronos failed to %s: %s", what, e) def _cancel(self, job_id: str) -> None: diff --git a/plugins/cron_providers/chronos/_nas_client.py b/plugins/cron_providers/chronos/_nas_client.py index 6777a271d5..2ef7843b5f 100644 --- a/plugins/cron_providers/chronos/_nas_client.py +++ b/plugins/cron_providers/chronos/_nas_client.py @@ -16,7 +16,22 @@ _LIST_PATH = "/api/agent-cron/list" class NasCronClientError(RuntimeError): - """Raised when a NAS agent-cron call fails (non-2xx or transport error).""" + """Raised when a NAS agent-cron call fails (non-2xx or transport error). + + ``status`` is the HTTP status (None on transport error) and ``error_code`` the OAuth-style + ``error`` field of a JSON error body (``invalid_client`` marks a deterministic identity + rejection the provider can act on, unlike a transient 5xx). + """ + + def __init__(self, message: str, *, status: int | None = None, error_code: str = "") -> None: + super().__init__(message) + self.status = status + self.error_code = error_code + + @property + def identity_rejected(self) -> bool: + """NAS refused the bearer as not belonging to a provisioned agent (never transient).""" + return self.status == 403 and self.error_code == "invalid_client" class NasCronClient: @@ -41,7 +56,12 @@ class NasCronClient: except Exception as e: raise NasCronClientError(f"{method} {path} failed: {e}") from e if resp.status_code // 100 != 2: - raise NasCronClientError(f"{method} {path} returned {resp.status_code}: {resp.text[:200]}") + error_code = "" + with contextlib.suppress(Exception): + error_code = str((resp.json() or {}).get("error") or "") + raise NasCronClientError( + f"{method} {path} returned {resp.status_code}: {resp.text[:200]}", + status=resp.status_code, error_code=error_code) with contextlib.suppress(Exception): return resp.json() if resp.content else {} return {} diff --git a/tests/cron/test_ticker_startup_survival.py b/tests/cron/test_ticker_startup_survival.py index 4884741228..ffb05da7d8 100644 --- a/tests/cron/test_ticker_startup_survival.py +++ b/tests/cron/test_ticker_startup_survival.py @@ -105,3 +105,17 @@ def test_housekeeping_restarts_a_dead_ticker(monkeypatch): stop.set() gateway_run._start_gateway_housekeeping(_OneTick(), interval=0, cron_thread=ticker) assert len(starts) == 2, "a stopped ticker must not be respawned" + + +def test_supervisor_leaves_a_returning_external_provider_alone(): + """An external provider's start() (Chronos) arms remote one-shots and RETURNS by design; the + supervisor must not read that as a dead ticker and re-run start() every housekeeping tick + (each rerun re-reconciles against NAS and logs a spurious "died without a stop request").""" + from cron.scheduler_thread import SupervisedTickerThread + + starts = [] + stop = threading.Event() + ticker = SupervisedTickerThread(lambda stop_event: starts.append(1), args=(stop,), stop_event=stop) + ticker.start() + _wait_until(lambda: not ticker.is_alive()) + assert ticker.restart_if_dead() is False and starts == [1] and ticker.restarts == 0 diff --git a/tests/plugins/test_chronos_cron.py b/tests/plugins/test_chronos_cron.py index 97486632bf..cb63781b97 100644 --- a/tests/plugins/test_chronos_cron.py +++ b/tests/plugins/test_chronos_cron.py @@ -100,6 +100,72 @@ def test_register_job_propagates_provision_failure(chronos): # -- reconcile ---------------------------------------------------------------- +def test_identity_rejection_hands_fires_to_the_builtin_ticker(temp_home, chronos, monkeypatch, caplog): + """Regression for #97494: NAS answering 403 invalid_client is a deterministic identity rejection + (the auth.json token is not this instance's provisioned agent), not a transient. The provider + must say so ONCE with the remedy, stop calling NAS, and start the built-in ticker so jobs still + fire on time instead of only via the late misfire sweep.""" + import threading + + from plugins.cron_providers.chronos._nas_client import NasCronClientError + + prov, fake = chronos + calls = [] + + def rejected(**kw): + calls.append(kw["job_id"]) + raise NasCronClientError( + "POST /api/agent-cron/provision returned 403: invalid_client", + status=403, error_code="invalid_client") + + fake.provision = rejected + jobs = [ + {"id": "a", "enabled": True, "next_run_at": "2026-06-18T12:00:00+00:00", "state": "scheduled"}, + {"id": "b", "enabled": True, "next_run_at": "2026-06-18T12:05:00+00:00", "state": "scheduled"}, + ] + monkeypatch.setattr("cron.jobs.load_jobs", lambda: jobs) + monkeypatch.setattr("cron.jobs.get_job", lambda jid: next(j for j in jobs if j["id"] == jid)) + monkeypatch.setattr("cron.executions.recover_interrupted_executions", lambda: 0) + ticker_started = threading.Event() + monkeypatch.setattr( + "cron.scheduler_provider.InProcessCronScheduler.start", + lambda self, stop_event, **kw: ticker_started.set()) + + stop = threading.Event() + with caplog.at_level("WARNING", logger="cron.chronos"): + prov.start(stop, adapters={"x": 1}, loop=None, interval=7) + assert ticker_started.wait(2.0), "the built-in ticker must take over this process's fires" + assert calls == ["a"], "after the first rejection NAS is left alone" + identity_msgs = [r.message for r in caplog.records if "re-login" in r.message] + assert len(identity_msgs) == 1 and "built-in cron ticker" in identity_msgs[0] + + # Job creation and re-arms no longer reach NAS (and no longer fail the create). + prov.register_job({"id": "c", "next_run_at": "2026-06-18T12:10:00+00:00"}) + prov.on_jobs_changed() + assert calls == ["a"] + + +def test_transient_provision_failure_does_not_degrade(temp_home, chronos, monkeypatch): + """A 5xx / transport error is retried on the next reconcile; only 403 invalid_client degrades.""" + from plugins.cron_providers.chronos._nas_client import NasCronClientError + + prov, fake = chronos + attempts = [] + + def flaky(**kw): + attempts.append(kw["job_id"]) + raise NasCronClientError("POST /api/agent-cron/provision returned 502: upstream", status=502) + + fake.provision = flaky + jobs = [{"id": "a", "enabled": True, "next_run_at": "2026-06-18T12:00:00+00:00", "state": "scheduled"}] + monkeypatch.setattr("cron.jobs.load_jobs", lambda: jobs) + monkeypatch.setattr("cron.jobs.get_job", lambda jid: jobs[0]) + + prov.reconcile() + prov.reconcile() + assert attempts == ["a", "a"] and prov._identity_rejected is False + + def test_reconcile_arms_all_enabled(temp_home, chronos, monkeypatch): prov, fake = chronos jobs = [ diff --git a/website/docs/developer-guide/chronos-managed-cron-contract.md b/website/docs/developer-guide/chronos-managed-cron-contract.md index fe331fea11..c7d4c34f40 100644 --- a/website/docs/developer-guide/chronos-managed-cron-contract.md +++ b/website/docs/developer-guide/chronos-managed-cron-contract.md @@ -219,6 +219,20 @@ If `callback_url` / `portal_url` is blank or the agent has no Nous login, `is_available()` returns False and the resolver falls back to the built-in in-process ticker — cron never loses its trigger. +**Identity rejection at runtime (`403 invalid_client`).** `is_available()` is +config-only, so it cannot tell whether the stored Nous token is the identity NAS +maps to a provisioned instance (hop 1 above). When `provision` answers 403 +`invalid_client` — the token in `auth.json` is a plain `hermes-cli` user login +rather than the `hermes-cli-vps` bootstrap session or an `agent:*` client — the +rejection is deterministic for the life of that credential: every arm, re-arm +and `list` would fail the same way, and a `hermes auth` re-login makes it +permanent (it *replaces* the bootstrap session; only NAS can re-mint one). The +provider therefore logs ONE warning naming that remedy, stops calling NAS, and +starts the built-in ticker for the rest of the process so jobs keep firing on +time instead of only through the late misfire sweep +(`cron.misfire_grace_minutes`). Transient failures (5xx, transport) do not +degrade; the next reconcile retries them. + ## Escape hatch (not default) The inbound `/api/cron/fire` verifier is pluggable (`get_fire_verifier()`). If