0.29.1 — review-fix round on E3: RunOutput, sshKeyPath fallback, RunStreaming consolidation, signer cache, stderr in errors

internal/httpapi/actuator.go

@@ -1,7 +1,6 @@
 package httpapi
 
 import (
-	"bytes"
 	"context"
 	"encoding/base64"
 	"encoding/json"
@@ -10,7 +9,6 @@ import (
 	"os"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/dtoro/oikos/internal/actuator"
@@ -77,24 +75,8 @@ const sshExecTimeout = 10 * time.Minute
 // streamWriter buffers everything it is given while forwarding each write to a
 // sink. One on session.Stdout and another sharing the same buffer on
 // session.Stderr reproduces CombinedOutput's interleaving in the order the
-// remote end produced it. Mirrors the twin in internal/mcp/server.go.
-type streamWriter struct {
-	mu     *sync.Mutex
-	buf    *bytes.Buffer
-	stream string
-	sink   execlog.Sink
-}
-
-func (w *streamWriter) Write(p []byte) (int, error) {
-	w.mu.Lock()
-	w.buf.Write(p)
-	w.mu.Unlock()
-	if w.sink != nil {
-		// Copy: the ssh library reuses p once Write returns.
-		w.sink(w.stream, append([]byte(nil), p...))
-	}
-	return len(p), nil
-}
+// remote end produced it. Shared implementation lives in internal/actuator
+// (actuator.streamWriter / actuator.RunStreaming).
 
 func sshExec(ctx context.Context, host, user, command string) (string, error) {
 	return sshExecStream(ctx, host, user, command, nil)
@@ -122,69 +104,7 @@ func sshExecStream(ctx context.Context, host, user, command string, sink execlog
 	}
 	defer client.Close()
 
-	session, err := client.NewSession()
-	if err != nil {
-		return "", fmt.Errorf("session: %w", err)
-	}
-	defer session.Close()
-
-	var (
-		mu  sync.Mutex
-		buf bytes.Buffer
-	)
-	session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink}
-	session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink}
-
-	collected := func() string {
-		mu.Lock()
-		defer mu.Unlock()
-		return strings.TrimSpace(buf.String())
-	}
-
-	done := make(chan error, 1)
-	go func() {
-		// See internal/mcp/server.go's sshExec for why this recovers rather
-		// than letting a rare SSH-library panic crash the whole api process.
-		defer func() {
-			if r := recover(); r != nil {
-				done <- fmt.Errorf("panic in ssh exec: %v", r)
-			}
-		}()
-		// Run rather than CombinedOutput so the assigned writers are used;
-		// Run returns only after both streams are fully drained.
-		done <- session.Run(command)
-	}()
-
-	select {
-	case err := <-done:
-		text := collected()
-		// A non-zero exit MUST surface as an error. The previous guard only
-		// errored when there was no output, so a `pct create` that printed
-		// "CT 132 already exists" and exited non-zero was reported as
-		// success — the execution was marked completed though nothing was
-		// provisioned.
-		if err != nil {
-			if text != "" {
-				return text, fmt.Errorf("%w: %s", err, text)
-			}
-			return text, fmt.Errorf("exec: %w", err)
-		}
-		return text, nil
-	case <-time.After(sshExecTimeout):
-		// Close the session/client to hang up the remote side; the
-		// goroutine above will eventually exit once that unblocks Run, but we
-		// don't wait for it — the caller needs an answer now, not an
-		// indefinite hang.
-		session.Close()
-		client.Close()
-		// Return what arrived before it hung, rather than "". A provisioning
-		// command that stalls halfway is precisely when its output matters.
-		return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
-	case <-ctx.Done():
-		session.Close()
-		client.Close()
-		return collected(), ctx.Err()
-	}
+	return actuator.RunStreaming(ctx, client, command, sink, sshExecTimeout)
 }
 
 func resolveHostSSH(ctx context.Context, pool *db.Pool, entitySlug string) (string, string, error) {

internal/mcp/server.go

@@ -452,30 +452,6 @@ func initSSH() {
 // goroutine forever with no way for the caller to ever get an answer.
 const sshExecTimeout = 10 * time.Minute
 
-// streamWriter buffers everything it is given while forwarding each write to a
-// sink. Assigning one to session.Stdout and another (sharing the same buffer)
-// to session.Stderr reproduces CombinedOutput's interleaving exactly, in the
-// order the remote end actually produced it — which reading from StdoutPipe
-// and StderrPipe separately would not guarantee.
-type streamWriter struct {
-	mu     *sync.Mutex
-	buf    *bytes.Buffer
-	stream string
-	sink   execlog.Sink
-}
-
-func (w *streamWriter) Write(p []byte) (int, error) {
-	w.mu.Lock()
-	w.buf.Write(p)
-	w.mu.Unlock()
-	if w.sink != nil {
-		// Copy: the ssh library reuses p after Write returns, and the sink
-		// hands the bytes to a DB call that may outlive this frame.
-		w.sink(w.stream, append([]byte(nil), p...))
-	}
-	return len(p), nil
-}
-
 func sshExec(ctx context.Context, host, user, command string) (string, error) {
 	return sshExecStream(ctx, host, user, command, nil)
 }
@@ -502,77 +478,7 @@ func sshExecStream(ctx context.Context, host, user, command string, sink execlog
 	}
 	defer client.Close()
 
-	session, err := client.NewSession()
-	if err != nil {
-		return "", fmt.Errorf("session: %w", err)
-	}
-	defer session.Close()
-
-	var (
-		mu  sync.Mutex
-		buf bytes.Buffer
-	)
-	session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink}
-	session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink}
-
-	// collected returns whatever output has arrived so far. Callable while the
-	// command is still running, which is what makes partial output on timeout
-	// possible.
-	collected := func() string {
-		mu.Lock()
-		defer mu.Unlock()
-		return strings.TrimSpace(buf.String())
-	}
-
-	done := make(chan error, 1)
-	go func() {
-		// Recovers a panic in the SSH library internals (rare but not
-		// impossible) and reports it as a failed command instead of crashing
-		// the whole api process — every gated action runs through this
-		// function, so an unrecovered panic here would take down every
-		// concurrently-running task's execution, not just this one. Without
-		// this, a panic would ALSO silently degrade to "wait out the full
-		// timeout" (done never receives, the select below falls through to
-		// its time.After case) rather than crashing outright — recovering
-		// and sending an immediate result is strictly better: the caller
-		// finds out now, not after sshExecTimeout.
-		defer func() {
-			if r := recover(); r != nil {
-				done <- fmt.Errorf("panic in ssh exec: %v", r)
-			}
-		}()
-		// Run rather than CombinedOutput so the assigned writers are used;
-		// Run returns only after both streams have been fully drained.
-		done <- session.Run(command)
-	}()
-
-	select {
-	case err := <-done:
-		text := collected()
-		// A non-zero exit MUST surface as an error — matching the fix
-		// applied to httpapi's sshExec (this copy still had the original
-		// bug: only erroring when there was no output at all, so a command
-		// that failed but printed something was silently reported as
-		// success).
-		if err != nil {
-			if text != "" {
-				return text, fmt.Errorf("%w: %s", err, text)
-			}
-			return text, fmt.Errorf("exec: %w", err)
-		}
-		return text, nil
-	case <-time.After(sshExecTimeout):
-		session.Close()
-		client.Close()
-		// Return what the command managed to print before it hung. This used
-		// to return "", discarding everything — so a hung command, the case
-		// where the output matters most, was the one case that left no trace.
-		return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
-	case <-ctx.Done():
-		session.Close()
-		client.Close()
-		return collected(), ctx.Err()
-	}
+	return actuator.RunStreaming(ctx, client, command, sink, sshExecTimeout)
 }
 
 // resolveHost resolves a host:<slug> to its reachable IP and SSH user. A thin

internal/mcp/tools.go

@@ -43,8 +43,6 @@ func formatCreateResult(slug, entityType string, res checkdefaults.Result) strin
 	return fmt.Sprintf("Created %s (%s).%s", slug, entityType, formatCheckResult(res))
 }
 
-// rowsToMap runs a SELECT key, value query and returns the result as a
-// map[string]any. Used by get_dashboard_summary to aggregate count queries.
 func rowsToMap(ctx context.Context, pool *db.Pool, query string, args ...any) map[string]any {
 	m := map[string]any{}
 	rows, err := pool.Query(ctx, query, args...)
@@ -62,8 +60,6 @@ func rowsToMap(ctx context.Context, pool *db.Pool, query string, args ...any) ma
 	return m
 }
 
-// queryRowsJSONSingle runs a query and returns the rows as a parsed JSON array
-// of maps. Used by get_ontology to embed sub-queries into a structured result.
 func queryRowsJSONSingle(ctx context.Context, pool *db.Pool, query string, args ...any) []map[string]any {
 	rows, err := pool.Query(ctx, query, args...)
 	if err != nil {

internal/scheduler/scheduler.go

@@ -7,11 +7,11 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"log/slog"
 	"net"
 	"net/http"
+	"os"
 	"os/exec"
 	"regexp"
 	"runtime"
@@ -19,6 +19,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/dtoro/oikos/internal/actuator"
 	"github.com/dtoro/oikos/internal/config"
 	"github.com/dtoro/oikos/internal/db"
 	"github.com/dtoro/oikos/internal/db/sqlcgen"
@@ -981,28 +982,46 @@ func allowlistedScript(name string) bool {
 	return scriptNameRe.MatchString(name)
 }
 
+// sshExec runs a command on a remote host over crypto/ssh via the shared
+// actuator primitives. It replaced a fork of `os/exec ssh` so the scheduler,
+// the MCP execution path, and the actuator share one dial/run/host-key
+// implementation (plan E3). The host key is verified through the centralized
+// actuator.HostKeyCallback seam. ctx bounds the running command; timeout
+// bounds the dial.
 func sshExec(ctx context.Context, host, port, user, cmd string, timeout time.Duration) ([]byte, error) {
-	args := []string{
-		"-o", "ConnectTimeout=" + strconv.Itoa(int(timeout.Seconds())),
-		"-o", "StrictHostKeyChecking=no",
-		"-o", "BatchMode=yes",
-		"-o", "UserKnownHostsFile=/dev/null",
-		"-o", "LogLevel=ERROR",
-	}
-	if sshKeyPath != "" {
-		args = append(args, "-i", sshKeyPath)
-	}
-	if port != "" && port != "22" {
-		args = append(args, "-p", port)
-	}
-	args = append(args, "-l", user, host, cmd)
-	c := exec.CommandContext(ctx, "ssh", args...)
-	out, err := c.Output()
-	if err != nil {
-		var ee *exec.ExitError
-		if errors.As(err, &ee) {
-			return nil, fmt.Errorf("ssh %s: %v (stderr: %s)", host, err, string(ee.Stderr))
+	keyPath := sshKeyPath
+	if keyPath == "" {
+		// Preserve the old os/exec-ssh behavior of deferring to a default
+		// key when no explicit OIKOS_SSH_KEY_PATH is configured: the system
+		// ssh binary used the agent / ~/.ssh; crypto/ssh has no agent wiring,
+		// so fall back to SSH_KEY_PATH then ~/.ssh/id_rsa.
+		keyPath = os.Getenv("SSH_KEY_PATH")
+		if keyPath == "" {
+			keyPath = os.Getenv("HOME") + "/.ssh/id_rsa"
 		}
+	}
+	signer, err := actuator.LoadSigner(keyPath)
+	if err != nil {
+		return nil, fmt.Errorf("ssh %s: %v", host, err)
+	}
+	p := 22
+	if port != "" {
+		if n, parseErr := strconv.Atoi(port); parseErr == nil && n > 0 {
+			p = n
+		}
+	}
+	client, err := actuator.Dial(ctx, actuator.DialOptions{
+		Host: host, Port: p, User: user, Signer: signer, Timeout: timeout,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("ssh %s: %v", host, err)
+	}
+	defer client.Close()
+	// RunOutput (stdout-only) — the scheduler parses check output as JSON or
+	// matches it literally, so stderr must not be merged in (RunCombinedOutput
+	// is for the live-run display path in mcp/httpapi).
+	out, err := actuator.RunOutput(ctx, client, cmd)
+	if err != nil {
 		return nil, fmt.Errorf("ssh %s: %v", host, err)
 	}
 	return out, nil