E1: split monolithic files — cmd/nomos (main.go → server.go + mcp.go + workers.go),
internal/mcp/tools.go → entity_tools/ops_tools/knowledge_tools/analysis_tools,
internal/httpapi/impl.go → domain files (entities, events, signals, ontology,
fleet_health, client_context, client_lifecycle, entity_mutations, query_audit).
E2: migrate raw pool.Exec queries to sqlc (entities/relationships queries + generated).
E3: unify SSH — consolidate crypto/ssh dial into actuator/client.go (+client_test).
E4/E5: add tests — db/lifecycle, checkdefaults/build, ontology/preconditions, policy/risk.
246 lines
6.8 KiB
Go
246 lines
6.8 KiB
Go
// Package actuator provides SSH-based skill procedure execution for the Oikos
|
|
// Phase 3 actuator loop. It runs stored skill procedures over SSH with a
|
|
// restricted key, classifies SSH errors into retryable/fatal/timeout, and
|
|
// supports step-by-step procedure verification.
|
|
package actuator
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"net"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"golang.org/x/crypto/ssh"
|
|
)
|
|
|
|
// ─── Procedure types ──────────────────────────────────────────────────────
|
|
|
|
// Procedure represents a parsed skill procedure from JSON config.
|
|
type Procedure struct {
|
|
Steps []Step `json:"steps"`
|
|
}
|
|
|
|
// Step is a single step within a procedure.
|
|
type Step struct {
|
|
Runner string `json:"runner"` // "shell", "script", "verify"
|
|
Target string `json:"target,omitempty"` // hostname/IP (empty = local)
|
|
Command string `json:"command"` // shell command or script path
|
|
TimeoutS int `json:"timeout_s,omitempty"` // per-step timeout in seconds
|
|
}
|
|
|
|
// SSHResult holds the outcome of an SSH execution.
|
|
type SSHResult struct {
|
|
Output string `json:"output"`
|
|
Duration time.Duration `json:"duration"`
|
|
Verified bool `json:"verified"`
|
|
Err error `json:"error,omitempty"`
|
|
}
|
|
|
|
// ─── Error classification ─────────────────────────────────────────────────
|
|
|
|
// SSHErrorClass categorises SSH errors.
|
|
type SSHErrorClass int
|
|
|
|
const (
|
|
SSHErrorUnknown SSHErrorClass = iota
|
|
SSHErrorNetwork // dial/connect timeout — retryable
|
|
SSHErrorAuth // auth failure — fatal
|
|
SSHErrorTimeout // command timed out
|
|
SSHErrorRemote // remote command returned non-zero
|
|
SSHErrorOther // other non-retryable
|
|
)
|
|
|
|
func (c SSHErrorClass) String() string {
|
|
switch c {
|
|
case SSHErrorNetwork:
|
|
return "network"
|
|
case SSHErrorAuth:
|
|
return "auth"
|
|
case SSHErrorTimeout:
|
|
return "timed_out"
|
|
case SSHErrorRemote:
|
|
return "remote"
|
|
case SSHErrorOther:
|
|
return "other"
|
|
default:
|
|
return "unknown"
|
|
}
|
|
}
|
|
|
|
// classifySSHError maps an SSH error to a class for retry/fatal decisions.
|
|
func classifySSHError(err error) SSHErrorClass {
|
|
if err == nil {
|
|
return SSHErrorOther
|
|
}
|
|
|
|
if errors.Is(err, context.DeadlineExceeded) {
|
|
return SSHErrorTimeout
|
|
}
|
|
|
|
var netErr net.Error
|
|
if errors.As(err, &netErr) {
|
|
if netErr.Timeout() {
|
|
return SSHErrorNetwork
|
|
}
|
|
return SSHErrorNetwork
|
|
}
|
|
|
|
if strings.Contains(err.Error(), "unable to authenticate") ||
|
|
strings.Contains(err.Error(), "no supported methods remain") ||
|
|
strings.Contains(err.Error(), "ssh: handshake failed") ||
|
|
strings.Contains(err.Error(), "publickey") ||
|
|
strings.Contains(err.Error(), "permission denied") {
|
|
return SSHErrorAuth
|
|
}
|
|
|
|
var exitErr *ssh.ExitError
|
|
if errors.As(err, &exitErr) {
|
|
return SSHErrorRemote
|
|
}
|
|
|
|
return SSHErrorOther
|
|
}
|
|
|
|
// ─── SSH execution ────────────────────────────────────────────────────────
|
|
|
|
// SSHConfig holds connection parameters for SSH sessions.
|
|
type SSHConfig struct {
|
|
Host string
|
|
Port int
|
|
User string
|
|
KeyPath string
|
|
Timeout time.Duration
|
|
}
|
|
|
|
// ExecuteProcedure runs a complete procedure over SSH, step by step.
|
|
// Returns the combined result, duration, and verified status.
|
|
//
|
|
// Context cancellation aborts the running session. Returns the last
|
|
// successfully completed step's output on partial failure.
|
|
func ExecuteProcedure(
|
|
ctx context.Context,
|
|
cfg SSHConfig,
|
|
proc Procedure,
|
|
) SSHResult {
|
|
start := time.Now()
|
|
|
|
// Parse the SSH key
|
|
signer, err := LoadSigner(cfg.KeyPath)
|
|
if err != nil {
|
|
return SSHResult{
|
|
Err: err,
|
|
Duration: time.Since(start),
|
|
Verified: false,
|
|
}
|
|
}
|
|
|
|
client, err := Dial(ctx, DialOptions{
|
|
Host: cfg.Host, Port: cfg.Port, User: cfg.User,
|
|
Signer: signer, Timeout: cfg.Timeout,
|
|
})
|
|
if err != nil {
|
|
class := classifySSHError(err)
|
|
return SSHResult{
|
|
Err: fmt.Errorf("ssh dial (%s): %w", class, err),
|
|
Duration: time.Since(start),
|
|
Verified: false,
|
|
}
|
|
}
|
|
defer client.Close()
|
|
|
|
// Execute each step in sequence
|
|
var lastOutput string
|
|
verified := true
|
|
|
|
for i, step := range proc.Steps {
|
|
// Check context before each step
|
|
if ctx.Err() != nil {
|
|
return SSHResult{
|
|
Output: lastOutput,
|
|
Duration: time.Since(start),
|
|
Verified: false,
|
|
Err: fmt.Errorf("cancelled before step %d: %w", i, ctx.Err()),
|
|
}
|
|
}
|
|
|
|
timeout := time.Duration(step.TimeoutS) * time.Second
|
|
if timeout <= 0 {
|
|
timeout = 30 * time.Second
|
|
}
|
|
|
|
stepCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
output, err := runSSHCommand(stepCtx, client, step.Command)
|
|
if err != nil {
|
|
class := classifySSHError(err)
|
|
// Verify steps that fail are not counted as verified failures
|
|
if step.Runner == "verify" {
|
|
verified = false
|
|
}
|
|
|
|
// Non-verify step failure is a real failure
|
|
if step.Runner != "verify" {
|
|
return SSHResult{
|
|
Output: lastOutput,
|
|
Duration: time.Since(start),
|
|
Err: fmt.Errorf("step %d (%s) failed (%s): %w", i, step.Runner, class, err),
|
|
Verified: false,
|
|
}
|
|
}
|
|
}
|
|
lastOutput = output
|
|
|
|
slog.Debug("ssh step completed",
|
|
"step", i,
|
|
"runner", step.Runner,
|
|
"duration", time.Since(start).Round(time.Millisecond),
|
|
)
|
|
}
|
|
|
|
return SSHResult{
|
|
Output: lastOutput,
|
|
Duration: time.Since(start),
|
|
Verified: verified,
|
|
}
|
|
}
|
|
|
|
// runSSHCommand executes a single command over an established SSH session via
|
|
// the shared RunCombinedOutput primitive (context-aware abort + combined output).
|
|
func runSSHCommand(ctx context.Context, client *ssh.Client, command string) (string, error) {
|
|
out, err := RunCombinedOutput(ctx, client, command)
|
|
return string(out), err
|
|
}
|
|
|
|
// ─── Procedure parsing ────────────────────────────────────────────────────
|
|
|
|
// ParseProcedure deserialises a JSON procedure (from skill.procedure).
|
|
func ParseProcedure(data []byte) (Procedure, error) {
|
|
var proc Procedure
|
|
if err := json.Unmarshal(data, &proc); err != nil {
|
|
return Procedure{}, fmt.Errorf("parse procedure: %w", err)
|
|
}
|
|
return proc, nil
|
|
}
|
|
|
|
// ─── Global SSH client options ────────────────────────────────────────────
|
|
|
|
var (
|
|
mu sync.Mutex
|
|
// defaultSSHTimeout is the default dial timeout for SSH connections.
|
|
defaultSSHTimeout = 10 * time.Second
|
|
)
|
|
|
|
// SetDefaultSSHTimeout overrides the default SSH dial timeout. Not safe for
|
|
// concurrent use during active execution.
|
|
func SetDefaultSSHTimeout(d time.Duration) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
defaultSSHTimeout = d
|
|
}
|