Compare commits

...

1 Commits

Author SHA1 Message Date
19369e273a fix(cron): validate prompt local service refs at job creation (#378)
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m14s
Shift-left: detect localhost/Ollama references in cron job prompts
at create/update time, not at runtime. Logs a warning with
actionable suggestion (set provider or base_url).

Changes:
- cron/jobs.py: Add import re, _LOCAL_SERVICE_CHECK_PATTERNS,
  _validate_job_prompt_local_refs(). Call from create_job()
  and update_job() when prompt is set.

Closes #378
2026-04-14 01:50:43 +00:00

View File

@@ -363,6 +363,45 @@ def save_jobs(jobs: List[Dict[str, Any]]):
raise
# Patterns that reference local services unreachable on cloud endpoints
_LOCAL_SERVICE_CHECK_PATTERNS = [
re.compile(r"\b(?:check|verify)\s+(?:that\s+)?ollama\b", re.IGNORECASE),
re.compile(r"\bcurl\s+(?:localhost|127\.0\.0\.1)", re.IGNORECASE),
re.compile(r"\bpoll\s+localhost\b", re.IGNORECASE),
re.compile(r"\bping\s+localhost\b", re.IGNORECASE),
re.compile(r"localhost:\d+", re.IGNORECASE),
re.compile(r"127\.0\.0\.1:\d+", re.IGNORECASE),
]
def _validate_job_prompt_local_refs(prompt: str, base_url: Optional[str] = None) -> List[str]:
"""Check if a cron job prompt references local services.
Returns list of warning messages (empty = no issues).
Warnings are advisory — jobs are NOT rejected.
"""
warnings = []
if not prompt:
return warnings
refs = []
for pat in _LOCAL_SERVICE_CHECK_PATTERNS:
found = pat.findall(prompt)
if found:
refs.extend(found[:2])
if refs:
refs_str = ", ".join(f"'{r}'" for r in refs[:5])
warnings.append(
f"Prompt references local services ({refs_str}) which may be "
f"unreachable if the job runs on a cloud provider. "
f"Consider setting provider='ollama' or base_url='http://localhost:11434/v1'."
)
return warnings
def create_job(
prompt: str,
schedule: str,
@@ -457,6 +496,11 @@ def create_job(
"origin": origin, # Tracks where job was created for "origin" delivery
}
# Validate prompt for local service references (#378)
_warnings = _validate_job_prompt_local_refs(prompt, normalized_base_url)
for w in _warnings:
logging.getLogger("cron.jobs").warning("Job '%s': %s", job_id, w)
jobs = load_jobs()
jobs.append(job)
save_jobs(jobs)