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")) }