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

@@ -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:

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 {}

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

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 = [

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