path: root/internal/executor/local.go

package executor

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/thepeterstone/claudomator/internal/llm"
	"github.com/thepeterstone/claudomator/internal/storage"
	"github.com/thepeterstone/claudomator/internal/task"
)

// LocalRunner executes a task against a local OpenAI-compatible LLM endpoint.
// Unlike ClaudeRunner/GeminiRunner it does not spawn a subprocess, does not
// create a git sandbox, and does not edit files in project_dir — it produces
// text completions that are streamed to stdout.log in the same stream-json
// envelope Claude uses, so existing parsers (extractSummary, ParseChangestat)
// keep working unchanged.
type LocalRunner struct {
	Client            *llm.Client
	Logger            *slog.Logger
	LogDir            string
	DefaultTemperature float64
}

// ExecLogDir implements LogPather so the pool can persist log paths before
// execution starts.
func (r *LocalRunner) ExecLogDir(execID string) string {
	if r.LogDir == "" {
		return ""
	}
	return filepath.Join(r.LogDir, execID)
}

// maxLocalToolTurns bounds the tool-use loop so a misbehaving model cannot spin
// forever calling tools without finishing.
const maxLocalToolTurns = 12

// Run drives a chat completion against the local endpoint with the agent
// back-channel tools (ask_user/report_summary/spawn_subtask/record_progress)
// declared. It loops, feeding tool results back as message history, until the
// model stops calling tools. Assistant text is written to stdout.log in the
// same stream-json envelope Claude uses so downstream parsers keep working.
// If the model calls ask_user, the run stops and returns a *BlockedError.
func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error {
	if r.Client == nil {
		return fmt.Errorf("local runner: no LLM client configured")
	}
	if t.Agent.Instructions == "" {
		return fmt.Errorf("local runner: empty instructions")
	}

	logDir := r.ExecLogDir(e.ID)
	if logDir == "" {
		return fmt.Errorf("local runner: LogDir not set")
	}
	if err := os.MkdirAll(logDir, 0o700); err != nil {
		return fmt.Errorf("local runner: mkdir log: %w", err)
	}
	stdoutPath := filepath.Join(logDir, "stdout.log")
	stderrPath := filepath.Join(logDir, "stderr.log")
	e.StdoutPath = stdoutPath
	e.StderrPath = stderrPath

	stdout, err := os.Create(stdoutPath)
	if err != nil {
		return fmt.Errorf("local runner: create stdout: %w", err)
	}
	defer stdout.Close()

	temperature := t.Agent.Temperature
	if temperature == nil && r.DefaultTemperature > 0 {
		v := r.DefaultTemperature
		temperature = &v
	}

	// Only provide tools if we aren't using a tiny model known to lack support.
	effectiveModel := t.Agent.Model
	if effectiveModel == "" && r.Client != nil {
		effectiveModel = r.Client.Model
	}
	var tools []llm.Tool
	if !strings.Contains(strings.ToLower(effectiveModel), "tinyllama") {
		tools = agentToolDefs()
	}

	// Build messages after tools are decided so the planning preamble is only
	// included when the model actually supports the agent tools.
	messages := []llm.Message{}
	if sys := strings.TrimSpace(t.Agent.SystemPromptAppend); sys != "" {
		messages = append(messages, llm.Message{Role: "system", Content: sys})
	}
	messages = append(messages, llm.Message{Role: "user", Content: buildAgentInstructions(t, len(tools) > 0)})

	start := time.Now()
	var totalIn, totalOut int
	var lastModel, lastFinish string
	blocked := false
	var fullText string

	for turn := 0; turn < maxLocalToolTurns; turn++ {
		req := llm.ChatRequest{
			Model:       t.Agent.Model,
			Messages:    messages,
			Temperature: temperature,
			MaxTokens:   t.Agent.MaxTokens,
			Tools:       tools,
		}
		resp, chatErr := r.Client.Chat(ctx, req)
		if chatErr != nil {
			// If the model doesn't support tool-use, retry once without tools.
			if tools != nil && strings.Contains(strings.ToLower(chatErr.Error()), "does not support tools") {
				tools = nil
				req.Tools = nil
				resp, chatErr = r.Client.Chat(ctx, req)
			}
			if chatErr != nil {
				writeResultLine(stdout, "error", chatErr.Error(), totalIn, totalOut)
				return fmt.Errorf("local runner: chat: %w", chatErr)
			}
		}
		totalIn += resp.PromptTokens
		totalOut += resp.OutputTokens
		lastModel, lastFinish = resp.Model, resp.FinishReason

		if resp.Content != "" {
			writeAssistantTextLine(stdout, resp.Content)
			fullText += resp.Content
		}

		if len(resp.ToolCalls) == 0 {
			break
		}

		// Re-feed: record the assistant's tool-call turn, then each tool result.
		messages = append(messages, llm.Message{Role: "assistant", Content: resp.Content, ToolCalls: resp.ToolCalls})
		for _, tc := range resp.ToolCalls {
			result, didBlock, toolErr := dispatchAgentTool(ctx, ch, tc.Function.Name, tc.Function.Arguments)
			if toolErr != nil {
				writeResultLine(stdout, "error", toolErr.Error(), totalIn, totalOut)
				return fmt.Errorf("local runner: tool %s: %w", tc.Function.Name, toolErr)
			}
			if didBlock {
				blocked = true
				break
			}
			messages = append(messages, llm.Message{
				Role:       "tool",
				ToolCallID: tc.ID,
				Name:       tc.Function.Name,
				Content:    result,
			})
		}
		if blocked {
			break
		}
	}
	elapsed := time.Since(start)

	e.CostUSD = 0
	e.TokensIn = int64(totalIn)
	e.TokensOut = int64(totalOut)

	if r.Logger != nil {
		r.Logger.Info("local runner completed",
			"taskID", t.ID,
			"model", lastModel,
			"tokens_in", totalIn,
			"tokens_out", totalOut,
			"finish_reason", lastFinish,
			"blocked", blocked,
			"elapsed_ms", elapsed.Milliseconds(),
		)
	}

	// If the model asked the user, stop and block. Local resume re-feeds the
	// conversation (per-runner sovereign state) — not yet wired, so the session
	// ID is empty and resume starts fresh for now.
	if q, isBlocked := channelPendingQuestion(ch); isBlocked {
		return &BlockedError{QuestionJSON: q}
	}

	// For tool-less models report_summary is never called, so synthesise a
	// summary from the raw assistant output so handleRunResult has something to
	// store.
	if e.Summary == "" && fullText != "" {
		s := strings.TrimSpace(fullText)
		if len(s) > 500 {
			s = s[:500]
		}
		e.Summary = s
	}

	writeResultLine(stdout, "success", "", totalIn, totalOut)
	return nil
}

// writeAssistantTextLine writes a single stream-json line wrapping `text` as
// an assistant text block. Format matches what ClaudeRunner emits, so
// extractSummary and ParseChangestatFromFile read it transparently.
func writeAssistantTextLine(w *os.File, text string) {
	line := struct {
		Type    string `json:"type"`
		Message struct {
			Content []struct {
				Type string `json:"type"`
				Text string `json:"text"`
			} `json:"content"`
		} `json:"message"`
	}{Type: "assistant"}
	line.Message.Content = []struct {
		Type string `json:"type"`
		Text string `json:"text"`
	}{{Type: "text", Text: text}}
	b, err := json.Marshal(line)
	if err != nil {
		return
	}
	w.Write(b)
	w.Write([]byte("\n"))
}

// writeResultLine writes a final stream-json terminator line that downstream
// parsers can recognise. Mirrors the shape of the result line ClaudeRunner emits.
func writeResultLine(w *os.File, subtype, errMsg string, promptTokens, outputTokens int) {
	line := map[string]any{
		"type":            "result",
		"subtype":         subtype,
		"is_error":        errMsg != "",
		"prompt_tokens":   promptTokens,
		"output_tokens":   outputTokens,
		"total_cost_usd":  0.0,
	}
	if errMsg != "" {
		line["result"] = errMsg
	}
	b, err := json.Marshal(line)
	if err != nil {
		return
	}
	w.Write(b)
	w.Write([]byte("\n"))
}