Timmy
57418dae07
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m16s
Cron jobs like nightwatch-health-monitor SSH into remote VPSes. When the runtime provider is cloud (Nous, OpenRouter, Anthropic), SSH keys don't exist on the inference server — causing silent failures and wasted iterations. Changes: - cron/scheduler.py: Import is_local_endpoint from model_metadata. Build disabled_toolsets dynamically: append 'terminal' when the runtime base_url is NOT a local endpoint. Log when terminal is disabled for observability. Also warn when a job declares requires_local_infra=true but runs on cloud. - tests/test_cron_cloud_terminal.py: 14 tests verifying is_local_endpoint classification and disabled_toolsets logic. Behavior: Local (localhost/127/RFC-1918): terminal enabled, SSH works. Cloud (openrouter/nous/anthropic): terminal disabled, agent reports SSH unavailable instead of wasting iterations. Closes #379
74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
"""Tests for cron scheduler cloud-provider terminal disabling (#379).
|
|
|
|
When a cron job runs on a cloud inference endpoint (Nous, OpenRouter, etc.),
|
|
the terminal toolset must be disabled because SSH keys don't exist on cloud
|
|
servers. Only local endpoints (localhost, 127.0.0.1, RFC-1918) retain
|
|
terminal access.
|
|
"""
|
|
|
|
import pytest
|
|
from agent.model_metadata import is_local_endpoint
|
|
|
|
|
|
class TestIsLocalEndpoint:
|
|
"""Verify is_local_endpoint correctly classifies endpoints."""
|
|
|
|
def test_localhost(self):
|
|
assert is_local_endpoint("http://localhost:11434/v1") is True
|
|
|
|
def test_127_loopback(self):
|
|
assert is_local_endpoint("http://127.0.0.1:8080/v1") is True
|
|
|
|
def test_0_0_0_0(self):
|
|
assert is_local_endpoint("http://0.0.0.0:11434/v1") is True
|
|
|
|
def test_rfc1918_10(self):
|
|
assert is_local_endpoint("http://10.0.0.5:8080/v1") is True
|
|
|
|
def test_rfc1918_192(self):
|
|
assert is_local_endpoint("http://192.168.1.100:11434/v1") is True
|
|
|
|
def test_rfc1918_172(self):
|
|
assert is_local_endpoint("http://172.16.0.1:8080/v1") is True
|
|
|
|
def test_cloud_openrouter(self):
|
|
assert is_local_endpoint("https://openrouter.ai/api/v1") is False
|
|
|
|
def test_cloud_nous(self):
|
|
assert is_local_endpoint("https://inference-api.nousresearch.com/v1") is False
|
|
|
|
def test_cloud_anthropic(self):
|
|
assert is_local_endpoint("https://api.anthropic.com") is False
|
|
|
|
def test_empty_url(self):
|
|
assert is_local_endpoint("") is False
|
|
|
|
def test_none_url(self):
|
|
assert is_local_endpoint(None) is False
|
|
|
|
|
|
class TestCronDisabledToolsetsLogic:
|
|
"""Verify the disabled_toolsets logic matches scheduler expectations."""
|
|
|
|
def _build_disabled(self, base_url, job=None):
|
|
"""Mirror the scheduler's disabled_toolsets logic."""
|
|
from agent.model_metadata import is_local_endpoint
|
|
cron_disabled = ["cronjob", "messaging", "clarify"]
|
|
if not is_local_endpoint(base_url):
|
|
cron_disabled.append("terminal")
|
|
return cron_disabled
|
|
|
|
def test_local_keeps_terminal(self):
|
|
disabled = self._build_disabled("http://localhost:11434/v1")
|
|
assert "terminal" not in disabled
|
|
assert "cronjob" in disabled
|
|
|
|
def test_cloud_disables_terminal(self):
|
|
disabled = self._build_disabled("https://openrouter.ai/api/v1")
|
|
assert "terminal" in disabled
|
|
assert "cronjob" in disabled
|
|
|
|
def test_empty_url_disables_terminal(self):
|
|
disabled = self._build_disabled("")
|
|
assert "terminal" in disabled
|