diff --git a/docker/stage2-hook.sh b/docker/stage2-hook.sh index 899c8e86ac989..6165eedf19ffc 100755 --- a/docker/stage2-hook.sh +++ b/docker/stage2-hook.sh @@ -431,6 +431,33 @@ seed_one ".env" ".env.example" seed_one "config.yaml" "cli-config.yaml.example" seed_one "SOUL.md" "docker/SOUL.md" +# --- Ensure a gateway api_server key exists (loopback control plane) --- +# The gateway's aiohttp api_server refuses to start without a strong +# API_SERVER_KEY (>=16 chars; startup guard in gateway/platforms/api_server.py). +# Hosted deployments need that listener on loopback so the dashboard — the +# container's only public HTTP door — can forward Chronos cron fires into the +# GATEWAY process, where the live platform adapters (relay, E2EE) live. The +# cron-fire route itself is NAS-JWT-authed, not key-authed; the key gates the +# rest of the api_server surface. Generate once, persist in .env (mounted +# volume), never overwrite an operator-provided value. Loopback-only: the +# default bind host is 127.0.0.1 and the Fly service only exposes the +# dashboard's port, so this listener is never publicly reachable. +if [ -f "$HERMES_HOME/.env" ] && ! grep -q '^API_SERVER_KEY=..*' "$HERMES_HOME/.env" 2>/dev/null; then + if refuse_symlinked_path "append" "$HERMES_HOME/.env"; then + : + else + _gen_key=$(head -c 32 /dev/urandom | od -An -tx1 | tr -d ' \n') + if [ -n "$_gen_key" ]; then + # Drop an empty assignment line if the seed left one behind, then + # append the generated key. + sed -i '/^API_SERVER_KEY=$/d' "$HERMES_HOME/.env" 2>/dev/null || true + printf 'API_SERVER_KEY=%s\n' "$_gen_key" >> "$HERMES_HOME/.env" + echo "[stage2] Generated API_SERVER_KEY for the loopback gateway api_server" + fi + unset _gen_key + fi +fi + # .env holds API keys and secrets — restrict to owner-only access. Applied # unconditionally (not only on first-seed) so a host-mounted .env that was # created with a permissive umask gets tightened on every container start. diff --git a/docs/chronos-managed-cron-contract.md b/docs/chronos-managed-cron-contract.md index 4692ea73d471a..f62de0f80fcb5 100644 --- a/docs/chronos-managed-cron-contract.md +++ b/docs/chronos-managed-cron-contract.md @@ -126,12 +126,26 @@ Arm (or re-arm, idempotently) exactly one one-shot for a job. ## Inbound `POST /api/cron/fire` (NAS → agent) — agent side, already implemented -This is the agent endpoint NAS calls in Endpoint 3 step 3. Served by the -**dashboard app** (`hermes_cli/web_server.py`) — the agent's always-reachable -public HTTP surface on hosted deployments (the gateway may be idle/scaled down); -it is in `PUBLIC_API_PATHS` so the dashboard cookie gate lets the bearer-JWT -callback through to the verifier. (Also registered on the optional -`APIServerAdapter` for self-host API-server deployments.) The verifier is +This is the agent endpoint NAS calls in Endpoint 3 step 3. Two hops on hosted +deployments: + +1. **Dashboard app** (`hermes_cli/web_server.py`) — the agent's only public + HTTP surface (the Fly proxy exposes exactly one port, the dashboard's). It + is in `PUBLIC_API_PATHS` so the dashboard cookie gate lets the bearer-JWT + callback through to the verifier. The dashboard verifies the JWT, resolves + the job's profile, then **forwards** the fire to hop 2 on loopback with the + NAS bearer preserved — it does NOT execute the job itself. +2. **Gateway `APIServerAdapter`** (`gateway/platforms/api_server.py`, loopback + bind, default port 8642) — re-verifies the JWT (defense in depth) and runs + the job with the gateway's **live platform adapters**, which is what makes + delivery work for relay-fronted logical platforms and E2EE rooms (the + standalone send path can serve neither). Self-host API-server deployments + that expose the api_server directly hit hop 2 without hop 1. + +Gateway unreachable from hop 1 (scale-to-zero wake still booting, restart +window, api_server disabled) → the dashboard returns **503** and NAS retries +(non-2xx = retryable, below); the store CAS de-dupes the eventual double fire. +There is deliberately no in-dashboard execution fallback. The verifier is `plugins/cron/chronos/verify.py`. - **Auth:** `Authorization: Bearer `. The agent verifies: diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 0097c1e11bb4d..5ac28ed05d6e7 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -5849,10 +5849,28 @@ async def _handle_cron_fire(self, request: "web.Request") -> "web.Response": provider = resolve_cron_scheduler() loop = asyncio.get_running_loop() + # Live adapters for delivery parity with the built-in ticker + # (gateway/run.py passes runner.adapters to the in-process + # scheduler). Without them, _deliver_result cannot resolve a live + # transport, so E2EE platforms and relay-fronted logical platforms + # (whose only send path IS the live relay adapter — no native + # credential exists) fail with "platform 'X' not + # configured/enabled" on every external-provider fire even though + # the same job delivers fine under the built-in ticker. + runner = self.gateway_runner or request.app.get("gateway_runner") + if runner is None: + try: + from gateway.run import _gateway_runner_ref + + runner = _gateway_runner_ref() + except Exception: + runner = None + adapters = getattr(runner, "adapters", None) or None + # Fire in the background (202 immediately). fire_due claims via the # store CAS, so a retry while this is in flight is de-duped. task = asyncio.create_task( - asyncio.to_thread(provider.fire_due, job_id, adapters=None, loop=loop) + asyncio.to_thread(provider.fire_due, job_id, adapters=adapters, loop=loop) ) reservation["detached"] = True task.add_done_callback( diff --git a/gateway/run.py b/gateway/run.py index a274636e1b11f..d6ee2991f8553 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -7732,6 +7732,34 @@ def _restart_loop_guard_config(self) -> tuple: pass return max_restarts, window_seconds, max_gap_seconds + def _scale_to_zero_active_messaging_platforms(self) -> list: + """ENABLED platforms that count for the relay-only arm gate (D1/F6). + + Two filters, both load-bearing: + - enabled only: config.platforms is pre-seeded with disabled + placeholders for the full platform catalog (the F25 bug). + - MESSAGING only: non-messaging surfaces must not disarm scale-to-zero. + The api_server is a loopback listener force-enabled by the presence + of API_SERVER_KEY (which the Docker stage2 hook now generates for + every container, so hosted instances ALWAYS have it enabled) — it + holds no outbound socket and Chronos fires through it already reset + the idle clock. Counting it made messaging_is_relay_only_or_absent + False on every hosted instance, silently disarming the feature. + Mirrors the non-messaging exclusion set used for handoff eligibility + (see the `messaging_platforms` computation in _connect_platforms). + """ + if not self.config: + return [] + non_messaging = {Platform.LOCAL, Platform.API_SERVER, Platform.WEBHOOK} + try: + return [ + p + for p, pc in self.config.platforms.items() + if getattr(pc, "enabled", False) and p not in non_messaging + ] + except Exception: # noqa: BLE001 + return [] + def _scale_to_zero_should_arm(self) -> bool: """Whether to start the idle watcher (D1/D11/§3.4(1)).""" from gateway.relay import relay_wake_url @@ -7741,23 +7769,7 @@ def _scale_to_zero_should_arm(self) -> bool: should_arm, ) - try: - # Only ENABLED platforms count. `config.platforms` is pre-seeded with a - # disabled placeholder PlatformConfig for every KNOWN platform (telegram, - # discord, slack, …), so `.keys()` is the full ~20-entry catalog regardless - # of what this instance actually runs. Passing the bare keys made - # `messaging_is_relay_only_or_absent` see those placeholders as live - # direct-socket platforms and return False, so scale-to-zero NEVER armed on - # a real relay-only instance. Mirror the connect loop, which already gates on - # `platform_config.enabled` (see the `if not platform_config.enabled: continue` - # in the adapter-connect loop) — arm off the same notion of "active platform." - platforms = ( - [p for p, pc in self.config.platforms.items() if getattr(pc, "enabled", False)] - if self.config - else [] - ) - except Exception: # noqa: BLE001 - platforms = [] + platforms = self._scale_to_zero_active_messaging_platforms() try: wake_url = relay_wake_url() except Exception: # noqa: BLE001 @@ -7786,18 +7798,10 @@ def _log_scale_to_zero_not_armed_reason(self) -> None: enabled = scale_to_zero_enabled() if not enabled: return # not opted in — normal, stay quiet - try: - active = ( - [ - getattr(p, "value", p) - for p, pc in self.config.platforms.items() - if getattr(pc, "enabled", False) - ] - if self.config - else [] - ) - except Exception: # noqa: BLE001 - active = [] + active = [ + getattr(p, "value", p) + for p in self._scale_to_zero_active_messaging_platforms() + ] relay_only = messaging_is_relay_only_or_absent(active) try: wake_url = relay_wake_url() diff --git a/hermes_cli/web_routers/cron.py b/hermes_cli/web_routers/cron.py index f5f861e710f27..e92b3c5ce53de 100644 --- a/hermes_cli/web_routers/cron.py +++ b/hermes_cli/web_routers/cron.py @@ -43,6 +43,7 @@ _delete_cron_job_sync = late("_delete_cron_job_sync") _find_cron_job_profile = late("_find_cron_job_profile") _fire_cron_job_for_profile = late("_fire_cron_job_for_profile") +_forward_cron_fire_to_gateway = late("_forward_cron_fire_to_gateway") _call_cron_for_profile = late("_call_cron_for_profile") _raise_if_cron_registration_error = late("_raise_if_cron_registration_error") load_config = late("load_config") @@ -124,22 +125,28 @@ async def delete_cron_job(job_id: str, profile: Optional[str] = None): @router.post("/api/cron/fire") async def cron_fire_webhook(request: Request): - """Chronos managed-cron fire webhook (NAS -> agent). + """Chronos managed-cron fire webhook (NAS -> agent) — gateway forwarder. Authenticated by a short-lived NAS-minted JWT (verified by the pluggable Chronos fire-verifier), NOT the dashboard session cookie — so this path is in ``PUBLIC_API_PATHS`` to bypass the dashboard auth gate, and the JWT is - the real gate. This is the inbound half of scale-to-zero managed cron: NAS - POSTs here at fire time, the agent verifies, claims the job (store CAS, so - at-most-once across replicas / on a NAS retry), runs it, and re-arms the - next one-shot. - - Lives on the dashboard app (not the api_server adapter) because the - dashboard is the agent's always-reachable public HTTP surface on hosted - deployments; the gateway may be idle/scaled down. - - Returns 202 immediately and runs the job in the background so a long agent - turn never trips NAS's HTTP timeout. + the real gate. + + The dashboard is only the PUBLIC DOOR here (on hosted deployments the Fly + proxy exposes exactly one port, the dashboard's). Cron execution belongs + to the GATEWAY process, which owns the live platform adapters — required + for relay-fronted logical platforms (their only sender is the live relay + adapter) and E2EE rooms, neither of which the dashboard's standalone send + path can serve. So after verifying the JWT this handler FORWARDS the fire + to the gateway api_server's own ``/api/cron/fire`` on loopback and passes + the gateway's response through (the gateway re-verifies the JWT — defense + in depth, no new trust link). + + Gateway unreachable (scale-to-zero wake still booting, restart window, + api_server disabled) → 503, so NAS retries per the Chronos contract + (non-2xx = retryable). The store CAS claim de-dupes the eventual double + fire. Deliberately NO local-execution fallback: delivering from the wrong + process is worse than a delayed retry. """ from plugins.cron_providers.chronos.verify import get_fire_verifier @@ -173,12 +180,20 @@ async def cron_fire_webhook(request: Request): # does not retry a fire that is intentionally absent. return JSONResponse({"status": "gone", "job_id": job_id}, status_code=200) - # Run in the background; the store CAS claim inside fire_due de-dupes a - # NAS/scheduler retry that arrives while this is in flight. - asyncio.create_task( - asyncio.to_thread(_fire_cron_job_for_profile, profile, job_id) - ) - return JSONResponse({"status": "accepted", "job_id": job_id}, status_code=202) + forwarded = await _forward_cron_fire_to_gateway(profile, job_id, auth) + if forwarded is None: + return JSONResponse( + { + "error": "gateway unreachable; retry", + "job_id": job_id, + "profile": profile, + }, + status_code=503, + ) + status_code, gateway_body = forwarded + if isinstance(gateway_body, dict): + gateway_body.setdefault("job_id", job_id) + return JSONResponse(gateway_body, status_code=status_code) @router.get("/api/cron/blueprints") diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index beabd6aa1399f..5e378bdda7773 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -12166,13 +12166,15 @@ def _delete_cron_job_sync(job_id: str, profile: Optional[str] = None): def _fire_cron_job_for_profile(profile: str, job_id: str) -> bool: - """Run ONE due cron job end-to-end for ``profile`` via the resolved - scheduler provider's ``fire_due`` (store CAS claim + ``run_one_job``). - - Scope both cron storage and the runtime Hermes home so the job's store, - config, credentials, scripts, skills, and output all belong to the selected - profile. Runs with no live adapters; delivery falls back to the per-platform - send path. + """DEPRECATED — retained only until callers migrate; do not add new uses. + + Superseded by :func:`_forward_cron_fire_to_gateway`: cron fires must + execute in the GATEWAY process (which owns the live platform adapters), + not the dashboard. Executing here delivered through the standalone path + only, which cannot serve relay-fronted logical platforms (their only + sender is the live relay adapter — no native credential exists on the + box) or E2EE rooms. Kept temporarily because external callers may still + resolve it via the web_deps late-binding seam. """ _profile_name, home = _cron_profile_home(profile) from cron import jobs as cron_jobs @@ -12191,6 +12193,143 @@ def _fire_cron_job_for_profile(profile: str, job_id: str) -> bool: reset_hermes_home_override(token) +def _profile_env_value(home: Path, key: str) -> str: + """Best-effort read of one KEY=VALUE line from a profile's .env file.""" + try: + env_path = home / ".env" + if not env_path.is_file(): + return "" + for line in env_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + if k.strip() == key: + return v.strip().strip('"').strip("'") + except Exception: + pass + return "" + + +def _gateway_fire_endpoint(profile: str, home: Path) -> str: + """Resolve the loopback URL of the gateway api_server's cron-fire route. + + Port resolution mirrors gateway/config.py's api_server load order for the + TARGET profile: ``platforms.api_server.extra.port`` in the profile's + config.yaml, then ``API_SERVER_PORT`` (process env for the active profile, + the profile's own .env otherwise), then the adapter default 8642. The bind + host is the adapter's loopback default — the dashboard and gateway share a + network namespace in every supported deployment (same host process tree, + or the same container under s6). + + Multiplex mode (one gateway serving several profiles) exposes per-profile + mirrors under ``/p//…``, so a non-default profile routes through + the default gateway's port with that prefix; per-profile-gateway mode + (each profile its own process/port) uses the bare path on the profile's + own port. + """ + import os as _os + + port = 0 + try: + # Profile-scoped read through the CANONICAL loader (managed-scope + # overlay, ${ENV_VAR} expansion, profile pathing) — never a raw + # yaml.safe_load of config.yaml (tests/hermes_cli/ + # test_config_read_guard.py). The HERMES_HOME override scopes + # get_config_path() to the TARGET profile, same pattern the + # deprecated _fire_cron_job_for_profile used for its store scope. + from hermes_constants import ( + reset_hermes_home_override, + set_hermes_home_override, + ) + + token = set_hermes_home_override(str(home)) + try: + profile_cfg = load_config() + finally: + reset_hermes_home_override(token) + raw = cfg_get( + profile_cfg, "platforms", "api_server", "extra", "port", default=None + ) + if raw: + port = int(raw) + except Exception: + port = 0 + if not port: + raw = ( + _os.getenv("API_SERVER_PORT", "") + if profile == _cron_default_profile() + else _profile_env_value(home, "API_SERVER_PORT") + ) + try: + port = int(raw) if raw else 0 + except ValueError: + port = 0 + if not port: + port = 8642 + + multiplex = False + try: + cfg = load_config() + multiplex = bool(cfg_get(cfg, "gateway", "multiplex_profiles", default=False)) + env_flag = _os.getenv("GATEWAY_MULTIPLEX_PROFILES", "").strip().lower() + if env_flag in {"1", "true", "yes", "on"}: + multiplex = True + elif env_flag in {"0", "false", "no", "off"}: + multiplex = False + except Exception: + pass + + if multiplex and profile != "default": + return f"http://127.0.0.1:{port}/p/{profile}/api/cron/fire" + return f"http://127.0.0.1:{port}/api/cron/fire" + + +async def _forward_cron_fire_to_gateway( + profile: str, job_id: str, authorization: str +) -> Optional[Tuple[int, Dict[str, Any]]]: + """Forward a Chronos fire callback to the gateway api_server on loopback. + + The dashboard is the hosted deployment's only public HTTP door (Fly proxy + → internal_port 9119), but cron execution belongs to the GATEWAY process: + it owns the live platform adapters, so delivery works for relay-fronted + logical platforms and E2EE rooms — the standalone path the dashboard used + to run cannot serve either. This forwards the fire byte-preserved (same + job_id, same NAS bearer — the gateway re-verifies the JWT itself) and + passes the gateway's response through. + + Returns ``(status_code, body)`` from the gateway, or ``None`` when the + gateway is unreachable (not started yet after a scale-to-zero wake, + restarting, or api_server disabled) — the caller maps that to 503 so NAS + retries per the Chronos contract (non-2xx = retryable; the store CAS + de-dupes the eventual double fire). + """ + _profile_name, home = _cron_profile_home(profile) + url = _gateway_fire_endpoint(_profile_name, home) + import httpx + + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.post( + url, + json={"job_id": job_id}, + headers={"Authorization": authorization}, + ) + except Exception as exc: + _log.warning( + "cron fire forward to %s failed (%s: %s); returning 503 for NAS retry", + url, type(exc).__name__, exc, + ) + return None + try: + body = resp.json() + except Exception: + body = {"raw": (resp.text or "")[:500]} + if not isinstance(body, dict): + body = {"raw": body} + return resp.status_code, body + + # --------------------------------------------------------------------------- diff --git a/tests/gateway/test_cron_fire_webhook.py b/tests/gateway/test_cron_fire_webhook.py index b8f8a3d4cf175..f6a9f94a7f294 100644 --- a/tests/gateway/test_cron_fire_webhook.py +++ b/tests/gateway/test_cron_fire_webhook.py @@ -238,3 +238,93 @@ async def async_verifier(**kw): break await asyncio.sleep(0.01) assert spy.fired == ["async-ok"] + + +@pytest.mark.asyncio +async def test_fire_passes_live_adapters_to_provider(adapter, monkeypatch): + """The fire webhook must hand the gateway's live adapters to fire_due — + delivery parity with the built-in ticker (gateway/run.py passes + runner.adapters). Without them, relay-fronted logical platforms (whose + ONLY send path is the live relay adapter — no native credential exists on + the box) and E2EE platforms fail every external-provider fire with + "platform 'X' not configured/enabled" while the same job delivers fine + under the in-process ticker.""" + seen = {} + + class _AdapterSpyProvider: + def fire_due(self, job_id, *, adapters=None, loop=None): + seen["job_id"] = job_id + seen["adapters"] = adapters + seen["loop"] = loop + return True + + live_adapters = {"relay": object()} + runner = SimpleNamespace( + _draining=False, + _external_drain_active=False, + adapters=live_adapters, + ) + + monkeypatch.setattr( + "cron.scheduler_provider.resolve_cron_scheduler", + lambda: _AdapterSpyProvider(), + ) + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + + with patch("gateway.run._gateway_runner_ref", lambda: runner): + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + resp = await cli.post("/api/cron/fire", + headers={"Authorization": "Bearer good"}, + json={"job_id": "with-adapters"}) + assert resp.status == 202 + + for _ in range(50): + if seen: + break + await asyncio.sleep(0.01) + + assert seen.get("job_id") == "with-adapters" + assert seen.get("adapters") is live_adapters + assert seen.get("loop") is not None + + +@pytest.mark.asyncio +async def test_fire_without_runner_passes_none_adapters(adapter, monkeypatch): + """No gateway runner (standalone/edge case) → fire still works with + adapters=None, preserving the historical standalone delivery path.""" + seen = {} + + class _AdapterSpyProvider: + def fire_due(self, job_id, *, adapters=None, loop=None): + seen["job_id"] = job_id + seen["adapters"] = adapters + return True + + monkeypatch.setattr( + "cron.scheduler_provider.resolve_cron_scheduler", + lambda: _AdapterSpyProvider(), + ) + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + + with patch("gateway.run._gateway_runner_ref", lambda: None): + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + resp = await cli.post("/api/cron/fire", + headers={"Authorization": "Bearer good"}, + json={"job_id": "no-runner"}) + assert resp.status == 202 + + for _ in range(50): + if seen: + break + await asyncio.sleep(0.01) + + assert seen.get("job_id") == "no-runner" + assert seen.get("adapters") is None diff --git a/tests/gateway/test_scale_to_zero_watcher.py b/tests/gateway/test_scale_to_zero_watcher.py index 781db57c4b8c4..f5e5eedac7ade 100644 --- a/tests/gateway/test_scale_to_zero_watcher.py +++ b/tests/gateway/test_scale_to_zero_watcher.py @@ -249,6 +249,63 @@ async def test_self_suspend_noop_off_fly(monkeypatch): await r._scale_to_zero_self_suspend() assert called == [] + +# ── non-messaging platforms must not disarm (the api_server-key regression) ── +# +# The Docker stage2 hook now generates API_SERVER_KEY for every container, and +# key presence force-enables the api_server platform (gateway/config.py). The +# arm gate counted every enabled platform, so `api_server` (a loopback +# listener, not a messaging socket) made messaging_is_relay_only_or_absent +# False on EVERY hosted instance — silently disarming scale-to-zero. The gate +# must only count messaging platforms (excluding LOCAL/API_SERVER/WEBHOOK, +# mirroring _connect_platforms' messaging_platforms exclusion set). + + +def test_arm_true_with_api_server_enabled(monkeypatch): + from gateway.platforms.base import Platform + + r = _arm_runner( + monkeypatch, + { + Platform.RELAY: True, + Platform.API_SERVER: True, + Platform.TELEGRAM: False, + }, + ) + assert r._scale_to_zero_should_arm() is True + + +def test_arm_true_with_all_non_messaging_surfaces_enabled(monkeypatch): + from gateway.platforms.base import Platform + + r = _arm_runner( + monkeypatch, + { + Platform.RELAY: True, + Platform.API_SERVER: True, + Platform.WEBHOOK: True, + Platform.LOCAL: True, + }, + ) + assert r._scale_to_zero_should_arm() is True + + +def test_direct_platform_still_disarms_alongside_api_server(monkeypatch): + """The messaging-only filter must not over-broaden: a genuinely enabled + direct-socket platform still disarms even with api_server also enabled.""" + from gateway.platforms.base import Platform + + r = _arm_runner( + monkeypatch, + { + Platform.RELAY: True, + Platform.API_SERVER: True, + Platform.DISCORD: True, + }, + ) + assert r._scale_to_zero_should_arm() is False + + # ── supervised watchers must NOT count as live background work (staging bug) ── # # _spawn_supervised parks every permanent watcher task (session-expiry, kanban, diff --git a/tests/hermes_cli/test_cron_fire_dashboard.py b/tests/hermes_cli/test_cron_fire_dashboard.py index 8bdf5b5ffd285..9ec70d4ade6cb 100644 --- a/tests/hermes_cli/test_cron_fire_dashboard.py +++ b/tests/hermes_cli/test_cron_fire_dashboard.py @@ -111,3 +111,141 @@ def test_unknown_job_200_gone(monkeypatch): client.close() +def test_valid_fire_forwards_to_gateway(monkeypatch): + """The dashboard is the public door only: a verified fire is FORWARDED to + the gateway api_server (which owns the live adapters — relay/E2EE + delivery), with the NAS bearer preserved, and the gateway's response is + passed through. The deprecated in-dashboard execution must NOT run.""" + forwarded = [] + executed = [] + + async def fake_forward(profile, job_id, authorization): + forwarded.append((profile, job_id, authorization)) + return 202, {"status": "accepted", "job_id": job_id} + + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + monkeypatch.setattr(web_server, "_find_cron_job_profile", lambda jid: "default") + monkeypatch.setattr(web_server, "_forward_cron_fire_to_gateway", fake_forward) + monkeypatch.setattr(web_server, "_fire_cron_job_for_profile", + lambda p, j: executed.append((p, j))) + + client, pa, ph = _client(auth_required=False) + try: + resp = client.post("/api/cron/fire", + headers={"Authorization": "Bearer nas-jwt"}, + json={"job_id": "j1"}) + assert resp.status_code == 202 + assert resp.json().get("status") == "accepted" + assert forwarded == [("default", "j1", "Bearer nas-jwt")] + assert executed == [] # no local execution — gateway owns cron + finally: + _restore(pa, ph) + client.close() + + +def test_gateway_unreachable_503_for_nas_retry(monkeypatch): + """Gateway down (scale-to-zero wake window / restart) -> 503 so NAS + retries per the Chronos contract. Deliberately NO local-execution + fallback — delivering from the dashboard process cannot serve + relay-fronted or E2EE targets.""" + executed = [] + + async def fake_forward(profile, job_id, authorization): + return None # unreachable + + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + monkeypatch.setattr(web_server, "_find_cron_job_profile", lambda jid: "default") + monkeypatch.setattr(web_server, "_forward_cron_fire_to_gateway", fake_forward) + monkeypatch.setattr(web_server, "_fire_cron_job_for_profile", + lambda p, j: executed.append((p, j))) + + client, pa, ph = _client(auth_required=False) + try: + resp = client.post("/api/cron/fire", + headers={"Authorization": "Bearer nas-jwt"}, + json={"job_id": "j2"}) + assert resp.status_code == 503 + assert executed == [] + finally: + _restore(pa, ph) + client.close() + + +def test_gateway_error_status_passes_through(monkeypatch): + """A non-2xx gateway response (e.g. its own 401 on a replayed JWT) passes + through unchanged — the dashboard adds no interpretation of its own.""" + + async def fake_forward(profile, job_id, authorization): + return 401, {"error": "invalid fire token"} + + monkeypatch.setattr( + "plugins.cron_providers.chronos.verify.get_fire_verifier", + lambda: (lambda **kw: {"purpose": "cron_fire"}), + ) + monkeypatch.setattr(web_server, "_find_cron_job_profile", lambda jid: "default") + monkeypatch.setattr(web_server, "_forward_cron_fire_to_gateway", fake_forward) + + client, pa, ph = _client(auth_required=False) + try: + resp = client.post("/api/cron/fire", + headers={"Authorization": "Bearer nas-jwt"}, + json={"job_id": "j3"}) + assert resp.status_code == 401 + finally: + _restore(pa, ph) + client.close() + + +# ── _gateway_fire_endpoint URL resolution ──────────────────────────────── + + +def test_fire_endpoint_default_port(tmp_path, monkeypatch): + monkeypatch.delenv("API_SERVER_PORT", raising=False) + monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False) + monkeypatch.setattr(web_server, "load_config", lambda: {}) + url = web_server._gateway_fire_endpoint("default", tmp_path) + assert url == "http://127.0.0.1:8642/api/cron/fire" + + +def test_fire_endpoint_config_yaml_port_wins(tmp_path, monkeypatch): + """The profile config.yaml port (read via the CANONICAL load_config, per + the config-read guard) wins over the process env API_SERVER_PORT.""" + monkeypatch.setenv("API_SERVER_PORT", "9999") + monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False) + monkeypatch.setattr( + web_server, + "load_config", + lambda: {"platforms": {"api_server": {"extra": {"port": 8700}}}}, + ) + url = web_server._gateway_fire_endpoint("default", tmp_path) + assert url == "http://127.0.0.1:8700/api/cron/fire" + + +def test_fire_endpoint_profile_env_port(tmp_path, monkeypatch): + """A non-default profile reads API_SERVER_PORT from its own .env, not the + dashboard process env (per-profile-gateway topology).""" + monkeypatch.setenv("API_SERVER_PORT", "9999") # dashboard process env + monkeypatch.delenv("GATEWAY_MULTIPLEX_PROFILES", raising=False) + monkeypatch.setattr(web_server, "load_config", lambda: {}) + monkeypatch.setattr(web_server, "_cron_default_profile", lambda: "default") + (tmp_path / ".env").write_text("API_SERVER_PORT=8701\n", encoding="utf-8") + url = web_server._gateway_fire_endpoint("worker_alpha", tmp_path) + assert url == "http://127.0.0.1:8701/api/cron/fire" + + +def test_fire_endpoint_multiplex_profile_prefix(tmp_path, monkeypatch): + """Multiplex mode: a non-default profile routes through the default + gateway's port with the /p// prefix mirror.""" + monkeypatch.delenv("API_SERVER_PORT", raising=False) + monkeypatch.setenv("GATEWAY_MULTIPLEX_PROFILES", "1") + monkeypatch.setattr(web_server, "load_config", lambda: {}) + url = web_server._gateway_fire_endpoint("worker_alpha", tmp_path) + assert url == "http://127.0.0.1:8642/p/worker_alpha/api/cron/fire" + +