fix(cron): managed-cron fires execute in the gateway process (live adapters + dashboard forwarder) (#84339)
* fix(gateway): pass live adapters to cron fire webhook's fire_due
The Chronos fire webhook (/api/cron/fire) called
provider.fire_due(job_id, adapters=None, loop=loop), so every
externally-triggered fire delivered through the standalone path even
with a live gateway in-process. E2EE platforms and relay-fronted
logical platforms (whose ONLY send path is the live relay adapter — no
native credential exists on the box) failed every external fire with
"platform 'X' not configured/enabled", while the same job delivered
fine under the built-in ticker (gateway/run.py passes runner.adapters).
Resolve the runner (self.gateway_runner → app['gateway_runner'] →
_gateway_runner_ref(), the same chain the drain check uses) and forward
its adapters. No runner → adapters=None, preserving the historical
standalone path byte-identically.
Note: does not by itself fix Fly-hosted scale-to-zero deployments where
NAS's callback lands on the DASHBOARD process (internal_port 9119) —
_fire_cron_job_for_profile there has no gateway runner in-process. That
topology needs a separate fire handoff (design pending).
* fix(cron): dashboard forwards Chronos fires to the gateway (503 when unreachable)
The dashboard's /api/cron/fire executed cron jobs in the DASHBOARD
process via _fire_cron_job_for_profile with adapters=None. On hosted
deployments (Fly proxy exposes only the dashboard's port) that made
every managed-cron fire deliver through the standalone send path, which
cannot serve relay-fronted logical platforms (their only sender is the
live relay adapter in the gateway process — no native credential exists
on the box) or E2EE rooms. It also ran the whole agent turn inside the
dashboard: wrong process for memory/session ownership and fire-claim
attribution.
Restore the invariant that the GATEWAY owns cron execution:
- Dashboard route: after verifying the NAS JWT and resolving the job's
profile, FORWARD the fire to the gateway api_server's own
/api/cron/fire on loopback, NAS bearer preserved (the gateway
re-verifies the JWT — defense in depth, no new trust link), and pass
the gateway's response through. Gateway unreachable → 503 so NAS
retries per the Chronos contract (non-2xx = retryable; the store CAS
de-dupes the eventual double fire). Deliberately NO local-execution
fallback.
- Endpoint resolution mirrors gateway/config.py's api_server load order
per target profile (config.yaml extra.port → API_SERVER_PORT from
process env or the profile's .env → 8642), with /p/<profile>/ prefix
routing under multiplex.
- docker/stage2-hook.sh: generate a strong API_SERVER_KEY into .env on
first boot when absent (never overwrites an operator value), so the
loopback api_server passes its startup guard on hosted images. The
fire route itself is NAS-JWT-authed; the key gates the rest of the
api_server surface. The listener binds 127.0.0.1 by default and the
Fly service exposes only the dashboard port.
- _fire_cron_job_for_profile kept but deprecated (late-binding seam
compatibility); no route calls it.
- docs/chronos-managed-cron-contract.md: document the two-hop inbound
topology and the 503-retry semantics.
Depends on the previous commit (fire webhook passes live adapters to
fire_due) — together they make NAS→dashboard→gateway fires deliver over
relay end to end.
* fix(cron): read the profile api_server port via the canonical config loader
CI guard test_config_read_guard flagged the new _gateway_fire_endpoint
for a raw yaml.safe_load of the profile's config.yaml — the exact drift
class the guard exists to kill (raw reads miss the managed-scope
overlay, ${ENV_VAR} expansion, and root-model normalization).
Read through load_config() under a HERMES_HOME override scoped to the
target profile instead (the same pattern the deprecated
_fire_cron_job_for_profile uses for its store scope), and pull the port
with cfg_get. Test updated to stub load_config rather than write a raw
config.yaml.
* fix(gateway): only messaging platforms count for the scale-to-zero arm gate
The stage2 hook now generates API_SERVER_KEY for every Docker container,
and key presence force-enables the api_server platform. The scale-to-zero
arm gate counted every enabled platform, so the loopback api_server
listener made messaging_is_relay_only_or_absent False on every hosted
instance — silently disarming the feature (the not-armed log would show
enabled platforms=['relay','api_server']).
The arm gate and the not-armed logger now share one helper that filters
to enabled MESSAGING platforms, excluding LOCAL/API_SERVER/WEBHOOK —
the same non-messaging exclusion set _connect_platforms already uses.
A genuinely enabled direct-socket platform (Discord/Telegram) still
disarms. Two of the three new tests fail without this fix.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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 <NAS-minted JWT>`. The agent verifies:
|
||||
|
||||
@@ -5849,10 +5849,28 @@ class APIServerAdapter(BasePlatformAdapter):
|
||||
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(
|
||||
|
||||
+33
-29
@@ -7732,6 +7732,34 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
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 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
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 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
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()
|
||||
|
||||
@@ -43,6 +43,7 @@ _trigger_cron_job_sync = late("_trigger_cron_job_sync")
|
||||
_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.
|
||||
the real gate.
|
||||
|
||||
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.
|
||||
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).
|
||||
|
||||
Returns 202 immediately and runs the job in the background so a long agent
|
||||
turn never trips NAS's HTTP timeout.
|
||||
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")
|
||||
|
||||
+145
-6
@@ -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``).
|
||||
"""DEPRECATED — retained only until callers migrate; do not add new uses.
|
||||
|
||||
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.
|
||||
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/<profile>/…``, 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
|
||||
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -238,3 +238,93 @@ async def test_async_verifier_is_awaited(adapter, monkeypatch):
|
||||
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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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/<profile>/ 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"
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user