Compare commits
2 Commits
queue/378-
...
fix/457-ss
| Author | SHA1 | Date | |
|---|---|---|---|
| 66b0febdfb | |||
| 6d79bf7783 |
@@ -13,7 +13,6 @@ import concurrent.futures
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import re
|
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
@@ -183,6 +182,15 @@ _SCRIPT_FAILURE_PHRASES = (
|
|||||||
"exit status",
|
"exit status",
|
||||||
"non-zero exit",
|
"non-zero exit",
|
||||||
"did not complete",
|
"did not complete",
|
||||||
|
# SSH-specific failure patterns (#350)
|
||||||
|
"no such file or directory",
|
||||||
|
"command not found",
|
||||||
|
"hermes binary not found",
|
||||||
|
"hermes not found",
|
||||||
|
"ssh: connect to host",
|
||||||
|
"connection timed out",
|
||||||
|
"host key verification failed",
|
||||||
|
"no route to host",
|
||||||
"could not run",
|
"could not run",
|
||||||
"unable to execute",
|
"unable to execute",
|
||||||
"permission denied",
|
"permission denied",
|
||||||
@@ -644,56 +652,7 @@ def _build_job_prompt(job: dict) -> str:
|
|||||||
return "\n".join(parts)
|
return "\n".join(parts)
|
||||||
|
|
||||||
|
|
||||||
# Regex patterns for local service references that fail on cloud endpoints
|
def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
|
||||||
_CLOUD_INCOMPATIBLE_PATTERNS = [
|
|
||||||
(re.compile(r"\b[Cc]heck\s+(?:that\s+)?[Oo]llama\s+(?:is\s+)?(?:responding|running|up|available)", re.IGNORECASE),
|
|
||||||
"Verify system services are healthy using available tools"),
|
|
||||||
(re.compile(r"\b[Vv]erify\s+(?:that\s+)?[Oo]llama\s+(?:is\s+)?(?:responding|running|up)", re.IGNORECASE),
|
|
||||||
"Verify system services are healthy using available tools"),
|
|
||||||
(re.compile(r"\bcurl\s+localhost:\d+", re.IGNORECASE),
|
|
||||||
"use available tools to check service health"),
|
|
||||||
(re.compile(r"\bcurl\s+127\.0\.0\.1:\d+", re.IGNORECASE),
|
|
||||||
"use available tools to check service health"),
|
|
||||||
(re.compile(r"\bpoll\s+localhost", re.IGNORECASE),
|
|
||||||
"check service health via available tools"),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def _rewrite_cloud_incompatible_prompt(prompt: str, base_url: str) -> str:
|
|
||||||
"""Rewrite prompt instructions that assume local service access when running on cloud.
|
|
||||||
|
|
||||||
When a cron job runs on a cloud inference endpoint (Nous, OpenRouter, Anthropic),
|
|
||||||
instructions to "Check Ollama" or "curl localhost:11434" are impossible.
|
|
||||||
Instead of just warning, this rewrites the instruction to a cloud-compatible
|
|
||||||
equivalent that the agent can actually execute.
|
|
||||||
|
|
||||||
Returns the (possibly rewritten) prompt.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
from agent.model_metadata import is_local_endpoint
|
|
||||||
except ImportError:
|
|
||||||
return prompt
|
|
||||||
|
|
||||||
if is_local_endpoint(base_url or ""):
|
|
||||||
return prompt # Local — no rewrite needed
|
|
||||||
|
|
||||||
rewritten = prompt
|
|
||||||
for pattern, replacement in _CLOUD_INCOMPATIBLE_PATTERNS:
|
|
||||||
rewritten = pattern.sub(replacement, rewritten)
|
|
||||||
|
|
||||||
if rewritten != prompt:
|
|
||||||
rewritten = (
|
|
||||||
"[NOTE: Some instructions were adjusted for cloud execution. "
|
|
||||||
"Local service checks were rewritten to use available tools.]
|
|
||||||
|
|
||||||
"
|
|
||||||
+ rewritten
|
|
||||||
)
|
|
||||||
|
|
||||||
return rewritten
|
|
||||||
|
|
||||||
|
|
||||||
def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:(job: dict) -> tuple[bool, str, str, Optional[str]]:
|
|
||||||
"""
|
"""
|
||||||
Execute a single cron job.
|
Execute a single cron job.
|
||||||
|
|
||||||
|
|||||||
212
cron/ssh_dispatch.py
Normal file
212
cron/ssh_dispatch.py
Normal file
@@ -0,0 +1,212 @@
|
|||||||
|
"""
|
||||||
|
SSH dispatch utilities for cron jobs.
|
||||||
|
|
||||||
|
Provides validated remote execution so broken hermes binary paths
|
||||||
|
are caught before draining the dispatch queue.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
from cron.ssh_dispatch import SSHEnvironment, format_dispatch_report
|
||||||
|
|
||||||
|
ssh = SSHEnvironment(host="root@ezra", agent="allegro")
|
||||||
|
result = ssh.dispatch("cron tick")
|
||||||
|
if not result.success:
|
||||||
|
print(result.failure_reason)
|
||||||
|
"""
|
||||||
|
|
||||||
|
import subprocess
|
||||||
|
import shutil
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import List, Optional
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class DispatchResult:
|
||||||
|
"""Structured result of a remote command dispatch."""
|
||||||
|
host: str
|
||||||
|
command: str
|
||||||
|
success: bool
|
||||||
|
exit_code: Optional[int] = None
|
||||||
|
stdout: str = ""
|
||||||
|
stderr: str = ""
|
||||||
|
failure_reason: Optional[str] = None
|
||||||
|
duration_s: float = 0.0
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class SSHEnvironment:
|
||||||
|
"""Validates and dispatches commands to a remote host via SSH."""
|
||||||
|
|
||||||
|
host: str # e.g. "root@ezra" or "192.168.1.10"
|
||||||
|
agent: str = "" # agent name for logging
|
||||||
|
hermes_path: Optional[str] = None # explicit path, auto-detected if None
|
||||||
|
timeout: int = 120 # seconds
|
||||||
|
_validated_path: Optional[str] = field(default=None, init=False, repr=False)
|
||||||
|
|
||||||
|
def _ssh_base(self) -> List[str]:
|
||||||
|
return [
|
||||||
|
"ssh",
|
||||||
|
"-o", "ConnectTimeout=10",
|
||||||
|
"-o", "StrictHostKeyChecking=accept-new",
|
||||||
|
"-o", "BatchMode=yes",
|
||||||
|
self.host,
|
||||||
|
]
|
||||||
|
|
||||||
|
def _probe_remote_binary(self, candidate: str) -> bool:
|
||||||
|
"""Check if a hermes binary exists and is executable on the remote host."""
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
self._ssh_base() + [f"test -x {candidate}"],
|
||||||
|
capture_output=True, timeout=15,
|
||||||
|
)
|
||||||
|
return result.returncode == 0
|
||||||
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def detect_hermes_binary(self) -> Optional[str]:
|
||||||
|
"""Find a working hermes binary on the remote host."""
|
||||||
|
if self._validated_path:
|
||||||
|
return self._validated_path
|
||||||
|
|
||||||
|
candidates = []
|
||||||
|
if self.hermes_path:
|
||||||
|
candidates.append(self.hermes_path)
|
||||||
|
|
||||||
|
# Common locations
|
||||||
|
candidates.extend([
|
||||||
|
"hermes", # on PATH
|
||||||
|
"~/.local/bin/hermes",
|
||||||
|
"/usr/local/bin/hermes",
|
||||||
|
f"~/wizards/{self.agent}/venv/bin/hermes" if self.agent else "",
|
||||||
|
f"/root/wizards/{self.agent}/venv/bin/hermes" if self.agent else "",
|
||||||
|
])
|
||||||
|
candidates = [c for c in candidates if c]
|
||||||
|
|
||||||
|
for candidate in candidates:
|
||||||
|
if self._probe_remote_binary(candidate):
|
||||||
|
self._validated_path = candidate
|
||||||
|
return candidate
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
def dispatch(self, command: str, *, validate_binary: bool = True) -> DispatchResult:
|
||||||
|
"""Execute a command on the remote host."""
|
||||||
|
import time
|
||||||
|
start = time.monotonic()
|
||||||
|
|
||||||
|
if validate_binary:
|
||||||
|
binary = self.detect_hermes_binary()
|
||||||
|
if not binary:
|
||||||
|
return DispatchResult(
|
||||||
|
host=self.host,
|
||||||
|
command=command,
|
||||||
|
success=False,
|
||||||
|
failure_reason=f"No working hermes binary found on {self.host}",
|
||||||
|
duration_s=time.monotonic() - start,
|
||||||
|
)
|
||||||
|
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
self._ssh_base() + [command],
|
||||||
|
capture_output=True,
|
||||||
|
timeout=self.timeout,
|
||||||
|
)
|
||||||
|
duration = time.monotonic() - start
|
||||||
|
stdout = result.stdout.decode("utf-8", errors="replace")
|
||||||
|
stderr = result.stderr.decode("utf-8", errors="replace")
|
||||||
|
|
||||||
|
failure_reason = None
|
||||||
|
if result.returncode != 0:
|
||||||
|
failure_reason = _classify_ssh_error(stderr, result.returncode)
|
||||||
|
|
||||||
|
return DispatchResult(
|
||||||
|
host=self.host,
|
||||||
|
command=command,
|
||||||
|
success=result.returncode == 0,
|
||||||
|
exit_code=result.returncode,
|
||||||
|
stdout=stdout,
|
||||||
|
stderr=stderr,
|
||||||
|
failure_reason=failure_reason,
|
||||||
|
duration_s=duration,
|
||||||
|
)
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
return DispatchResult(
|
||||||
|
host=self.host,
|
||||||
|
command=command,
|
||||||
|
success=False,
|
||||||
|
failure_reason=f"SSH command timed out after {self.timeout}s",
|
||||||
|
duration_s=time.monotonic() - start,
|
||||||
|
)
|
||||||
|
except FileNotFoundError:
|
||||||
|
return DispatchResult(
|
||||||
|
host=self.host,
|
||||||
|
command=command,
|
||||||
|
success=False,
|
||||||
|
failure_reason="ssh binary not found on local system",
|
||||||
|
duration_s=time.monotonic() - start,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _classify_ssh_error(stderr: str, exit_code: int) -> str:
|
||||||
|
"""Classify an SSH error from stderr and exit code."""
|
||||||
|
lower = stderr.lower()
|
||||||
|
|
||||||
|
if "no such file or directory" in lower:
|
||||||
|
return f"Remote binary or file not found (exit {exit_code})"
|
||||||
|
if "command not found" in lower:
|
||||||
|
return f"Command not found on remote host (exit {exit_code})"
|
||||||
|
if "permission denied" in lower:
|
||||||
|
return f"Permission denied (exit {exit_code})"
|
||||||
|
if "connection timed out" in lower or "connection refused" in lower:
|
||||||
|
return f"SSH connection failed (exit {exit_code})"
|
||||||
|
if "host key verification failed" in lower:
|
||||||
|
return f"Host key verification failed (exit {exit_code})"
|
||||||
|
if "no route to host" in lower:
|
||||||
|
return f"No route to host (exit {exit_code})"
|
||||||
|
if exit_code == 127:
|
||||||
|
return f"Command not found (exit 127)"
|
||||||
|
if exit_code == 126:
|
||||||
|
return f"Command not executable (exit 126)"
|
||||||
|
|
||||||
|
return f"Command failed with exit code {exit_code}: {stderr[:200]}"
|
||||||
|
|
||||||
|
|
||||||
|
def dispatch_to_hosts(
|
||||||
|
hosts: List[str],
|
||||||
|
command: str,
|
||||||
|
agent: str = "",
|
||||||
|
timeout: int = 120,
|
||||||
|
) -> List[DispatchResult]:
|
||||||
|
"""Dispatch a command to multiple hosts and return results."""
|
||||||
|
results = []
|
||||||
|
for host in hosts:
|
||||||
|
ssh = SSHEnvironment(host=host, agent=agent, timeout=timeout)
|
||||||
|
result = ssh.dispatch(command)
|
||||||
|
results.append(result)
|
||||||
|
return results
|
||||||
|
|
||||||
|
|
||||||
|
def format_dispatch_report(results: List[DispatchResult]) -> str:
|
||||||
|
"""Format a human-readable report of dispatch results."""
|
||||||
|
lines = ["## Dispatch Report", ""]
|
||||||
|
|
||||||
|
succeeded = [r for r in results if r.success]
|
||||||
|
failed = [r for r in results if not r.success]
|
||||||
|
|
||||||
|
lines.append(f"**Total:** {len(results)} hosts | "
|
||||||
|
f"**OK:** {len(succeeded)} | **Failed:** {len(failed)}")
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
for r in results:
|
||||||
|
status = "OK" if r.success else "FAIL"
|
||||||
|
lines.append(f"### {r.host} [{status}]")
|
||||||
|
lines.append(f"- Command: `{r.command}`")
|
||||||
|
lines.append(f"- Duration: {r.duration_s:.1f}s")
|
||||||
|
if r.exit_code is not None:
|
||||||
|
lines.append(f"- Exit code: {r.exit_code}")
|
||||||
|
if r.failure_reason:
|
||||||
|
lines.append(f"- **Failure:** {r.failure_reason}")
|
||||||
|
if r.stderr and not r.success:
|
||||||
|
lines.append(f"- Stderr: `{r.stderr[:300]}`")
|
||||||
|
lines.append("")
|
||||||
|
|
||||||
|
return "\n".join(lines)
|
||||||
Reference in New Issue
Block a user