agents_and_robots/shell/effects/runner.go

// Package effects interprets pure []decision.Action values into real side effects.
package effects

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/enmanuel/agents/pkg/decision"
	"github.com/enmanuel/agents/pkg/tools/devicemesh"
	"github.com/enmanuel/agents/shell/logger"
	"github.com/enmanuel/agents/shell/ssh"
)

// DeviceMeshCaller is the minimal interface that the Runner needs from a
// devicemesh.ToolRegistry. It is an interface (rather than a concrete type)
// so tests can mock without spinning up an HTTP server.
type DeviceMeshCaller interface {
	Call(ctx context.Context, toolName string, input map[string]any) (any, error)
}

// Compile-time check: the real registry satisfies the interface.
var _ DeviceMeshCaller = (*devicemesh.ToolRegistry)(nil)

// Result holds the outcome of executing a single action.
type Result struct {
	Action decision.Action
	Output string
	Err    error
}

// MatrixSender is satisfied by shell/matrix.Client.
type MatrixSender interface {
	SendText(ctx context.Context, roomID, text string) error
	SendMarkdown(ctx context.Context, roomID, markdown string) error
	SendMarkdownGetID(ctx context.Context, roomID, markdown string) (string, error)
	EditMessage(ctx context.Context, roomID, originalEventID, markdown string) error
	SendReplyMarkdown(ctx context.Context, roomID, inReplyTo, markdown string) error
	SendThreadMarkdown(ctx context.Context, roomID, threadRootID, inReplyTo, markdown string) error
	SendTyping(ctx context.Context, roomID string, typing bool) error
}

// Runner interprets actions and executes them.
type Runner struct {
	matrix     MatrixSender
	ssh        *ssh.Executor
	deviceMesh DeviceMeshCaller
	logger     *slog.Logger
}

// NewRunner creates a Runner with the provided dependencies.
// The device mesh tool registry is left nil; ActionKindDeviceMesh actions
// will be rejected with a clear error. Use NewRunnerWithDeviceMesh to wire
// the mesh caller.
func NewRunner(matrix MatrixSender, ssh *ssh.Executor, logger *slog.Logger) *Runner {
	return &Runner{matrix: matrix, ssh: ssh, logger: logger}
}

// NewRunnerWithDeviceMesh wires a Runner with a DeviceMeshCaller, enabling
// ActionKindDeviceMesh dispatch. Used by the launcher when an agent has
// cfg.DeviceMesh.Enabled = true (wiring lives in 0144c).
func NewRunnerWithDeviceMesh(matrix MatrixSender, ssh *ssh.Executor, dm DeviceMeshCaller, logger *slog.Logger) *Runner {
	return &Runner{matrix: matrix, ssh: ssh, deviceMesh: dm, logger: logger}
}

// Execute runs each action sequentially and returns results.
func (r *Runner) Execute(ctx context.Context, roomID string, actions []decision.Action) []Result {
	r.logger.Debug("effects_batch", "room", roomID, "count", len(actions))
	results := make([]Result, 0, len(actions))
	for _, a := range actions {
		start := time.Now()
		res := r.executeOne(ctx, roomID, a)
		ms := time.Since(start).Milliseconds()
		results = append(results, res)
		if res.Err != nil {
			r.logger.Error("action_failed", logger.FieldAction, a.Kind, logger.FieldDurationMS, ms, "err", res.Err)
		} else {
			r.logger.Info("action_done", logger.FieldAction, a.Kind, logger.FieldDurationMS, ms)
		}
	}
	return results
}

func (r *Runner) executeOne(ctx context.Context, roomID string, a decision.Action) Result {
	switch a.Kind {
	case decision.ActionKindReply:
		if a.Reply == nil {
			return Result{Action: a, Err: fmt.Errorf("nil reply action")}
		}
		var err error
		switch {
		case a.Reply.ThreadID != "":
			// Thread reply: send as part of the thread with fallback in_reply_to
			err = r.matrix.SendThreadMarkdown(ctx, roomID, a.Reply.ThreadID, a.Reply.InReplyTo, a.Reply.Content)
		case a.Reply.InReplyTo != "":
			err = r.matrix.SendReplyMarkdown(ctx, roomID, a.Reply.InReplyTo, a.Reply.Content)
		default:
			err = r.matrix.SendMarkdown(ctx, roomID, a.Reply.Content)
		}
		return Result{Action: a, Output: a.Reply.Content, Err: err}

	case decision.ActionKindSSH:
		if a.SSH == nil {
			return Result{Action: a, Err: fmt.Errorf("nil ssh action")}
		}
		res := r.ssh.Execute(ctx, *a.SSH)
		output := res.Stdout
		if res.Stderr != "" {
			output += "\nstderr: " + res.Stderr
		}
		return Result{Action: a, Output: output, Err: res.Err}

	case decision.ActionKindDeviceMesh:
		if a.DeviceMesh == nil {
			return Result{Action: a, Err: fmt.Errorf("nil device_mesh action")}
		}
		if r.deviceMesh == nil {
			return Result{Action: a, Err: fmt.Errorf("device_mesh action received but Runner has no DeviceMeshCaller (build with NewRunnerWithDeviceMesh)")}
		}
		result, err := r.deviceMesh.Call(ctx, a.DeviceMesh.Tool, a.DeviceMesh.Input)
		output := formatDeviceMeshResult(result)
		return Result{Action: a, Output: output, Err: err}

	default:
		return Result{Action: a, Err: fmt.Errorf("unhandled action kind: %s", a.Kind)}
	}
}

// formatDeviceMeshResult renders the tool result as a stable JSON string
// suitable for embedding in a tool_result message to the LLM. Errors during
// marshaling collapse to a printable Go representation — never panic, never
// drop data on the floor.
func formatDeviceMeshResult(v any) string {
	if v == nil {
		return ""
	}
	if s, ok := v.(string); ok {
		return s
	}
	b, err := json.Marshal(v)
	if err != nil {
		return fmt.Sprintf("%v", v)
	}
	return string(b)
}