diff options
Diffstat (limited to 'internal/executor')
| -rw-r--r-- | internal/executor/channel.go | 51 | ||||
| -rw-r--r-- | internal/executor/helpers.go | 11 | ||||
| -rw-r--r-- | internal/executor/local.go | 282 | ||||
| -rw-r--r-- | internal/executor/localtools.go | 279 | ||||
| -rw-r--r-- | internal/executor/nativerunner.go | 127 | ||||
| -rw-r--r-- | internal/executor/nativerunner_test.go (renamed from internal/executor/local_test.go) | 160 |
6 files changed, 288 insertions, 622 deletions
diff --git a/internal/executor/channel.go b/internal/executor/channel.go index 8605ffa..debe21d 100644 --- a/internal/executor/channel.go +++ b/internal/executor/channel.go @@ -3,46 +3,33 @@ package executor import ( "context" "encoding/json" - "errors" "sync" "time" + "github.com/thepeterstone/claudomator/internal/agentchannel" "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/task" "github.com/google/uuid" ) -// ErrAgentBlocked is returned by AgentChannel.AskUser when the transport cannot -// deliver an answer within the current run — e.g. the file transport, which can -// only write-and-exit. Runners translate it into a *BlockedError so the task -// blocks and resumes later. Transports that can suspend the agent in-session -// (MCP, Phase 2) instead block and return the answer. -var ErrAgentBlocked = errors.New("agent blocked awaiting user input") - -// SubtaskSpec describes a child task an agent wants to spawn. -type SubtaskSpec struct { - Name string - Instructions string - Model string - MaxBudgetUSD float64 -} - -// AgentChannel is how a Runner reports agent-originated signals to the rest of -// the system. Implementations translate these into stored artifacts and events. -// The transport by which a Runner detects these signals — post-exit files -// today, MCP/tool-use in Phase 2 — is the Runner's private concern. -type AgentChannel interface { - // AskUser records that the agent needs human input. Transports that can - // suspend the agent in-session return the answer; those that cannot return - // ErrAgentBlocked. - AskUser(ctx context.Context, questionJSON string) (answer string, err error) - // ReportSummary records the agent's final summary text. - ReportSummary(ctx context.Context, summary string) error - // SpawnSubtask creates a child task and returns its ID. - SpawnSubtask(ctx context.Context, spec SubtaskSpec) (taskID string, err error) - // RecordProgress records a free-form progress note from the agent. - RecordProgress(ctx context.Context, message string) error -} +// AgentChannel, SubtaskSpec, BlockedError, and ErrAgentBlocked used to be +// defined directly in this package. Phase 1 of the multi-provider refactor +// moved them to internal/agentchannel (a small leaf package with no +// executor dependency) so that internal/agentloop — the new provider-neutral +// tool-use loop — can construct/inspect them without importing +// internal/executor. That matters because internal/executor now imports +// internal/agentloop (via nativerunner.go, which wires an agentloop.Loop up +// to a Runner); if agentloop imported executor too, that would be a direct +// import cycle. Re-exporting via type aliases (and a var assignment for the +// sentinel error) here means every existing reference to +// executor.AgentChannel / executor.SubtaskSpec / executor.BlockedError / +// executor.ErrAgentBlocked keeps compiling and behaving identically — an +// alias is the same type, not a copy. +type AgentChannel = agentchannel.AgentChannel +type SubtaskSpec = agentchannel.SubtaskSpec +type BlockedError = agentchannel.BlockedError + +var ErrAgentBlocked = agentchannel.ErrAgentBlocked // channelStore is the subset of storage the default channel needs. type channelStore interface { diff --git a/internal/executor/helpers.go b/internal/executor/helpers.go index f2111a4..f4dd02c 100644 --- a/internal/executor/helpers.go +++ b/internal/executor/helpers.go @@ -10,15 +10,8 @@ import ( "strings" ) -// BlockedError is returned by Run when the agent wrote a question file and exited. -// The pool transitions the task to BLOCKED and stores the question for the user. -type BlockedError struct { - QuestionJSON string // raw JSON from the question file - SessionID string // claude session to resume once the user answers - SandboxDir string // preserved sandbox path; resume must run here so Claude finds its session files -} - -func (e *BlockedError) Error() string { return fmt.Sprintf("task blocked: %s", e.QuestionJSON) } +// BlockedError is now an alias for agentchannel.BlockedError, defined in +// channel.go — see the comment there for the Phase 1 import-cycle rationale. // parseStream reads streaming JSON from claude, writes to w, and returns // (costUSD, error). error is non-nil if the stream signals task failure: diff --git a/internal/executor/local.go b/internal/executor/local.go deleted file mode 100644 index 43e76d3..0000000 --- a/internal/executor/local.go +++ /dev/null @@ -1,282 +0,0 @@ -package executor - -import ( - "context" - "encoding/json" - "fmt" - "log/slog" - "os" - "os/exec" - "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() - - // --- Git sandbox setup --- - var workDir string - if t.Agent.ProjectDir != "" { - // Reuse existing sandbox on resume (BLOCKED → QUEUED path). - if e.SandboxDir != "" { - workDir = e.SandboxDir - } else { - tmpDir, tmpErr := os.MkdirTemp("", "claudomator-local-*") - if tmpErr != nil { - return fmt.Errorf("local runner: create sandbox temp dir: %w", tmpErr) - } - cloneCmd := exec.CommandContext(ctx, "git", "clone", t.Agent.ProjectDir, tmpDir) - if cloneOut, cloneErr := cloneCmd.CombinedOutput(); cloneErr != nil { - os.RemoveAll(tmpDir) - return fmt.Errorf("local runner: git clone: %w\n%s", cloneErr, cloneOut) - } - workDir = tmpDir - e.SandboxDir = workDir - } - } - - 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, workDir, 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. Preserve the sandbox so - // resume can reuse it. - if q, isBlocked := channelPendingQuestion(ch); isBlocked { - // workDir is already stored in e.SandboxDir; don't clean up. - return &BlockedError{QuestionJSON: q} - } - - // Push any commits made inside the sandbox back to the origin (project_dir). - if workDir != "" { - pushCmd := exec.CommandContext(ctx, "git", "-C", workDir, "push", "origin", "HEAD") - if pushOut, pushErr := pushCmd.CombinedOutput(); pushErr != nil { - if r.Logger != nil { - r.Logger.Warn("local runner: git push failed", "err", pushErr, "output", string(pushOut)) - } - } - os.RemoveAll(workDir) - e.SandboxDir = "" - } - - // 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")) -} diff --git a/internal/executor/localtools.go b/internal/executor/localtools.go deleted file mode 100644 index 9fd2cfc..0000000 --- a/internal/executor/localtools.go +++ /dev/null @@ -1,279 +0,0 @@ -package executor - -import ( - "bytes" - "context" - "encoding/json" - "errors" - "fmt" - "os" - "os/exec" - "path/filepath" - - "github.com/thepeterstone/claudomator/internal/llm" -) - -// agentToolDefs returns the four agent back-channel tools as OpenAI -// function-calling definitions, for runners that talk to an OpenAI-compatible -// endpoint (LocalRunner). They mirror the MCP tools the container runners expose. -func agentToolDefs() []llm.Tool { - strProp := func(desc string) map[string]any { - return map[string]any{"type": "string", "description": desc} - } - return []llm.Tool{ - {Type: "function", Function: llm.ToolFunction{ - Name: "ask_user", - Description: "Ask the user a question when you genuinely need a decision to proceed. The task pauses until the user answers; do not call other tools after this.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{ - "question": strProp("the question to ask, phrased as a real question"), - "options": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "description": "optional suggested answer choices"}, - }, - "required": []string{"question"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "report_summary", - Description: "Record a concise 2-5 sentence summary of what you accomplished. Call this before finishing.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{"summary": strProp("the summary text")}, - "required": []string{"summary"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "spawn_subtask", - Description: "Create a child task to be executed separately. Use this to break large work into focused pieces.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{ - "name": strProp("short descriptive name for the subtask"), - "instructions": strProp("complete instructions for the subtask agent"), - "model": strProp("optional model override"), - "max_budget_usd": map[string]any{"type": "number", "description": "optional budget cap in USD"}, - }, - "required": []string{"name", "instructions"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "record_progress", - Description: "Record a short progress note that appears in the task timeline.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{"message": strProp("a short progress note")}, - "required": []string{"message"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "read_file", - Description: "Read the contents of a file in the sandbox working directory.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{"path": strProp("path to the file, relative or absolute")}, - "required": []string{"path"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "write_file", - Description: "Write or overwrite a file in the sandbox working directory.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{ - "path": strProp("path to the file, relative or absolute"), - "content": strProp("content to write"), - }, - "required": []string{"path", "content"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "run_bash", - Description: "Run a shell command in the sandbox working directory. Returns exit code, stdout, and stderr.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{"command": strProp("shell command to execute")}, - "required": []string{"command"}, - }, - }}, - {Type: "function", Function: llm.ToolFunction{ - Name: "glob", - Description: "List files matching a glob pattern relative to the sandbox working directory.", - Parameters: map[string]any{ - "type": "object", - "properties": map[string]any{"pattern": strProp("glob pattern, e.g. **/*.go")}, - "required": []string{"pattern"}, - }, - }}, - } -} - -// dispatchAgentTool invokes one model-requested tool against the AgentChannel. -// workDir is the sandbox working directory; file/bash tools return an error if -// it is empty. It returns the text to feed back to the model as the tool -// result, and a blocked flag set when ask_user could not be answered in-session -// (the run must stop and the task block). -func dispatchAgentTool(ctx context.Context, ch AgentChannel, workDir, name, argsJSON string) (result string, blocked bool, err error) { - switch name { - case "ask_user": - var a struct { - Question string `json:"question"` - Options []string `json:"options"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - q := map[string]any{"text": a.Question} - if len(a.Options) > 0 { - q["options"] = a.Options - } - payload, _ := json.Marshal(q) - ans, askErr := ch.AskUser(ctx, string(payload)) - if errors.Is(askErr, ErrAgentBlocked) { - return "", true, nil - } - if askErr != nil { - return "", false, askErr - } - return ans, false, nil - - case "report_summary": - var a struct { - Summary string `json:"summary"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - if rsErr := ch.ReportSummary(ctx, a.Summary); rsErr != nil { - return "", false, rsErr - } - return "Summary recorded.", false, nil - - case "spawn_subtask": - var a struct { - Name string `json:"name"` - Instructions string `json:"instructions"` - Model string `json:"model"` - MaxBudgetUSD float64 `json:"max_budget_usd"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - id, ssErr := ch.SpawnSubtask(ctx, SubtaskSpec{ - Name: a.Name, - Instructions: a.Instructions, - Model: a.Model, - MaxBudgetUSD: a.MaxBudgetUSD, - }) - if ssErr != nil { - return "", false, ssErr - } - return "Created subtask " + id, false, nil - - case "record_progress": - var a struct { - Message string `json:"message"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - if rpErr := ch.RecordProgress(ctx, a.Message); rpErr != nil { - return "", false, rpErr - } - return "Noted.", false, nil - - case "read_file": - if workDir == "" { - return "", false, fmt.Errorf("read_file: no sandbox working directory") - } - var a struct { - Path string `json:"path"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - p := a.Path - if !filepath.IsAbs(p) { - p = filepath.Join(workDir, p) - } - data, readErr := os.ReadFile(p) - if readErr != nil { - return "", false, readErr - } - return string(data), false, nil - - case "write_file": - if workDir == "" { - return "", false, fmt.Errorf("write_file: no sandbox working directory") - } - var a struct { - Path string `json:"path"` - Content string `json:"content"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - p := a.Path - if !filepath.IsAbs(p) { - p = filepath.Join(workDir, p) - } - if mkErr := os.MkdirAll(filepath.Dir(p), 0o755); mkErr != nil { - return "", false, mkErr - } - if writeErr := os.WriteFile(p, []byte(a.Content), 0o644); writeErr != nil { - return "", false, writeErr - } - return "Written.", false, nil - - case "run_bash": - if workDir == "" { - return "", false, fmt.Errorf("run_bash: no sandbox working directory") - } - var a struct { - Command string `json:"command"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - cmd := exec.CommandContext(ctx, "sh", "-c", a.Command) - cmd.Dir = workDir - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - runErr := cmd.Run() - exitCode := 0 - if runErr != nil { - if exitErr, ok := runErr.(*exec.ExitError); ok { - exitCode = exitErr.ExitCode() - } else { - return "", false, runErr - } - } - const maxOutput = 8 * 1024 - out := stdout.String() - errOut := stderr.String() - if len(out) > maxOutput { - out = out[:maxOutput] + "\n[truncated]" - } - if len(errOut) > maxOutput { - errOut = errOut[:maxOutput] + "\n[truncated]" - } - return fmt.Sprintf("exit_code: %d\nstdout:\n%s\nstderr:\n%s", exitCode, out, errOut), false, nil - - case "glob": - if workDir == "" { - return "", false, fmt.Errorf("glob: no sandbox working directory") - } - var a struct { - Pattern string `json:"pattern"` - } - _ = json.Unmarshal([]byte(argsJSON), &a) - pattern := a.Pattern - if !filepath.IsAbs(pattern) { - pattern = filepath.Join(workDir, pattern) - } - matches, globErr := filepath.Glob(pattern) - if globErr != nil { - return "", false, globErr - } - // Return paths relative to workDir for readability. - rel := make([]string, 0, len(matches)) - for _, m := range matches { - r, relErr := filepath.Rel(workDir, m) - if relErr != nil { - r = m - } - rel = append(rel, r) - } - out, _ := json.Marshal(rel) - return string(out), false, nil - - default: - return "", false, fmt.Errorf("unknown tool %q", name) - } -} diff --git a/internal/executor/nativerunner.go b/internal/executor/nativerunner.go new file mode 100644 index 0000000..6fe4196 --- /dev/null +++ b/internal/executor/nativerunner.go @@ -0,0 +1,127 @@ +package executor + +import ( + "context" + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" + + "github.com/thepeterstone/claudomator/internal/agentloop" + "github.com/thepeterstone/claudomator/internal/provider" + "github.com/thepeterstone/claudomator/internal/sandbox" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +// NativeRunner executes a task against a provider.Provider (a native chat/ +// tool-use backend for some LLM vendor's wire format) using the shared +// agentloop.Loop tool-use control flow. It supersedes LocalRunner, which drove +// the same control flow inline against internal/llm.Client directly; that +// logic was extracted into internal/agentloop (and internal/sandbox for the +// git/file/bash/glob tool implementations) so any provider.Provider — not +// just the OpenAI-compatible one — can share it. Like LocalRunner, it does +// not spawn a subprocess: it produces text/tool-call completions that are +// streamed to stdout.log in the same stream-json envelope ClaudeRunner uses, +// so existing parsers (extractSummary, task.ParseChangestatFromFile) keep +// working unchanged. +type NativeRunner struct { + Provider provider.Provider + Logger *slog.Logger + LogDir string + + // DefaultTemperature is used when a task doesn't set its own + // Agent.Temperature. + DefaultTemperature float64 + + // DefaultModel names the provider's default model (e.g. config.LocalModel's + // configured model). It's used only to decide whether agent tools should be + // declared at all (some tiny local models — "tinyllama" in the model name — + // don't support tool-use) when the task itself doesn't specify a model, + // mirroring LocalRunner's historical fallback to its llm.Client.Model. + DefaultModel string + + // MaxTurns bounds the tool-use loop; defaults to agentloop.DefaultMaxTurns + // (12, matching the former executor.maxLocalToolTurns) when zero. + MaxTurns int +} + +var _ Runner = (*NativeRunner)(nil) +var _ LogPather = (*NativeRunner)(nil) + +// ExecLogDir implements LogPather so the pool can persist log paths before +// execution starts. Identical to LocalRunner.ExecLogDir. +func (r *NativeRunner) ExecLogDir(execID string) string { + if r.LogDir == "" { + return "" + } + return filepath.Join(r.LogDir, execID) +} + +// Run drives a chat completion against r.Provider with the agent back-channel +// tools (ask_user/report_summary/spawn_subtask/record_progress) and sandbox +// tools (read_file/write_file/run_bash/glob) declared, via agentloop.Loop. If +// the model calls ask_user, the run stops and returns a *BlockedError. +func (r *NativeRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { + if r.Provider == nil { + return fmt.Errorf("native runner: no provider configured") + } + if t.Agent.Instructions == "" { + return fmt.Errorf("native runner: empty instructions") + } + + logDir := r.ExecLogDir(e.ID) + if logDir == "" { + return fmt.Errorf("native runner: LogDir not set") + } + if err := os.MkdirAll(logDir, 0o700); err != nil { + return fmt.Errorf("native 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("native runner: create stdout: %w", err) + } + defer stdout.Close() + + // --- Sandbox setup (git clone into a temp dir, or reuse on resume) --- + sb := sandbox.NewHostSandbox("") + if t.Agent.ProjectDir != "" { + if e.SandboxDir != "" { + // Reuse existing sandbox on resume (BLOCKED → QUEUED path). + sb = sandbox.NewHostSandbox(e.SandboxDir) + } else { + if cloneErr := sb.GitClone(ctx, t.Agent.ProjectDir, ""); cloneErr != nil { + return fmt.Errorf("native runner: %w", cloneErr) + } + e.SandboxDir = sb.WorkDir() + } + } + + // Only provide tools if we aren't using a tiny model known to lack support. + effectiveModel := t.Agent.Model + if effectiveModel == "" { + effectiveModel = r.DefaultModel + } + toolsEnabled := !strings.Contains(strings.ToLower(effectiveModel), "tinyllama") + + // Build the agent's instructions after the tools decision so the planning + // preamble is only included when the model actually supports the agent + // tools — same ordering LocalRunner used. + instructions := buildAgentInstructions(t, toolsEnabled) + + loop := &agentloop.Loop{ + Provider: r.Provider, + Sandbox: sb, + Channel: ch, + MaxTurns: r.MaxTurns, + DefaultTemperature: r.DefaultTemperature, + Logger: r.Logger, + } + return loop.Run(ctx, t, e, stdout, instructions, toolsEnabled) +} diff --git a/internal/executor/local_test.go b/internal/executor/nativerunner_test.go index 2ae8380..f76b365 100644 --- a/internal/executor/local_test.go +++ b/internal/executor/nativerunner_test.go @@ -9,6 +9,7 @@ import ( "net/http" "net/http/httptest" "os" + "os/exec" "path/filepath" "strings" "sync" @@ -16,10 +17,50 @@ import ( "github.com/google/uuid" "github.com/thepeterstone/claudomator/internal/llm" + "github.com/thepeterstone/claudomator/internal/provider/openaicompat" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" ) +// TestNativeRunner_Run_ChangestatsParseable confirms that a git diff --stat +// summary line embedded in the model's assistant text (the only way such a +// line can reach stdout.log — tool results themselves are never logged) is +// still recoverable by task.ParseChangestatFromFile via the new +// NativeRunner/agentloop path, exactly as it was via the old LocalRunner +// (neither runner logs raw tool output; both only log assistant text + +// result lines, so this has always depended on the model itself echoing the +// diff stat in its reply). +func TestNativeRunner_Run_ChangestatsParseable(t *testing.T) { + srv := fakeChatServer(t, []fakeTurn{ + {content: "## Summary\nCommitted the fix.\n2 files changed, 30 insertions(+), 4 deletions(-)", promptTok: 3, outTok: 7}, + }) + defer srv.Close() + + r := newLocalRunner(t, srv) + tt := localTask() + exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} + + if err := r.Run(context.Background(), tt, exec, newStoreChannel(&fakeChannelStore{}, tt.ID)); err != nil { + t.Fatalf("Run: %v", err) + } + + stdoutPath := filepath.Join(r.ExecLogDir(exec.ID), "stdout.log") + cs := task.ParseChangestatFromFile(stdoutPath) + if cs == nil { + t.Fatalf("ParseChangestatFromFile found nothing in %s", stdoutPath) + } + if cs.FilesChanged != 2 || cs.LinesAdded != 30 || cs.LinesRemoved != 4 { + t.Errorf("changestats: got %+v, want {2 30 4}", cs) + } +} + +// This file exercises NativeRunner — the provider.Provider/agentloop.Loop +// based successor to the former LocalRunner — via a fake OpenAI-compatible +// HTTP server, using the same fake-server test pattern LocalRunner's tests +// used (fakeChatServer/fakeTurn/toolCall below), so it verifies behavioral +// equivalence: same stdout.log stream-json shape, same summary/changestats +// extraction, same tool-loop semantics. + // fakeTurn is one canned chat-completion response the fake server returns. type fakeTurn struct { content string @@ -60,12 +101,13 @@ func toolCall(id, name, args string) llm.ToolCall { return llm.ToolCall{ID: id, Type: "function", Function: llm.ToolCallFunction{Name: name, Arguments: args}} } -func newLocalRunner(t *testing.T, srv *httptest.Server) *LocalRunner { +func newLocalRunner(t *testing.T, srv *httptest.Server) *NativeRunner { t.Helper() - return &LocalRunner{ - Client: &llm.Client{Endpoint: srv.URL + "/v1", Model: "fake"}, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: t.TempDir(), + client := &llm.Client{Endpoint: srv.URL + "/v1", Model: "fake"} + return &NativeRunner{ + Provider: openaicompat.New(client), + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + LogDir: t.TempDir(), } } @@ -82,7 +124,7 @@ func localTask() *task.Task { } } -func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { +func TestNativeRunner_Run_WritesStreamJSON(t *testing.T) { srv := fakeChatServer(t, []fakeTurn{{content: "## Summary\nAll good.", promptTok: 11, outTok: 22}}) defer srv.Close() @@ -95,7 +137,7 @@ func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { } if exec.CostUSD != 0 { - t.Errorf("CostUSD should be 0 for local runner, got %v", exec.CostUSD) + t.Errorf("CostUSD should be 0 for native runner, got %v", exec.CostUSD) } if exec.TokensIn != 11 || exec.TokensOut != 22 { t.Errorf("tokens: want 11/22 got %d/%d", exec.TokensIn, exec.TokensOut) @@ -123,7 +165,7 @@ func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { } } -func TestLocalRunner_Run_ToolLoop_ReportSummaryAndSpawn(t *testing.T) { +func TestNativeRunner_Run_ToolLoop_ReportSummaryAndSpawn(t *testing.T) { // Turn 1: model spawns a subtask. Turn 2: reports summary. Turn 3: finishes. srv := fakeChatServer(t, []fakeTurn{ {toolCalls: []llm.ToolCall{toolCall("c1", "spawn_subtask", `{"name":"sub A","instructions":"do sub"}`)}, promptTok: 5, outTok: 3}, @@ -157,7 +199,7 @@ func TestLocalRunner_Run_ToolLoop_ReportSummaryAndSpawn(t *testing.T) { } } -func TestLocalRunner_Run_RecordProgress(t *testing.T) { +func TestNativeRunner_Run_RecordProgress(t *testing.T) { srv := fakeChatServer(t, []fakeTurn{ {toolCalls: []llm.ToolCall{toolCall("c1", "record_progress", `{"message":"halfway there"}`)}}, {content: "done"}, @@ -178,7 +220,7 @@ func TestLocalRunner_Run_RecordProgress(t *testing.T) { } } -func TestLocalRunner_Run_AskUser_Blocks(t *testing.T) { +func TestNativeRunner_Run_AskUser_Blocks(t *testing.T) { srv := fakeChatServer(t, []fakeTurn{ {toolCalls: []llm.ToolCall{toolCall("c1", "ask_user", `{"question":"which branch?"}`)}}, {content: "should not be reached"}, @@ -198,20 +240,20 @@ func TestLocalRunner_Run_AskUser_Blocks(t *testing.T) { } } -func TestLocalRunner_Run_NoClient_Errors(t *testing.T) { - r := &LocalRunner{LogDir: t.TempDir()} +func TestNativeRunner_Run_NoProvider_Errors(t *testing.T) { + r := &NativeRunner{LogDir: t.TempDir()} tt := &task.Task{ID: "x", Agent: task.AgentConfig{Instructions: "hi"}} exec := &storage.Execution{ID: "exec-x"} err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)) - if err == nil || !strings.Contains(err.Error(), "no LLM client") { - t.Errorf("expected 'no LLM client' error, got %v", err) + if err == nil || !strings.Contains(err.Error(), "no provider configured") { + t.Errorf("expected 'no provider configured' error, got %v", err) } } -func TestLocalRunner_Run_EmptyInstructions_Errors(t *testing.T) { - r := &LocalRunner{ - Client: &llm.Client{Endpoint: "http://unused", Model: "x"}, - LogDir: t.TempDir(), +func TestNativeRunner_Run_EmptyInstructions_Errors(t *testing.T) { + r := &NativeRunner{ + Provider: openaicompat.New(&llm.Client{Endpoint: "http://unused", Model: "x"}), + LogDir: t.TempDir(), } tt := &task.Task{ID: "x", Agent: task.AgentConfig{}} exec := &storage.Execution{ID: "exec-x"} @@ -221,8 +263,8 @@ func TestLocalRunner_Run_EmptyInstructions_Errors(t *testing.T) { } } -func TestLocalRunner_ExecLogDir(t *testing.T) { - r := &LocalRunner{LogDir: "/tmp/logs"} +func TestNativeRunner_ExecLogDir(t *testing.T) { + r := &NativeRunner{LogDir: "/tmp/logs"} if got := r.ExecLogDir("abc"); got != "/tmp/logs/abc" { t.Errorf("ExecLogDir: got %q", got) } @@ -231,3 +273,81 @@ func TestLocalRunner_ExecLogDir(t *testing.T) { t.Errorf("ExecLogDir empty LogDir: got %q", got) } } + +// TestNativeRunner_Run_ProjectDir_SandboxToolsAndPush drives a task with +// Agent.ProjectDir set through write_file/read_file/glob/run_bash tool calls +// against a real HostSandbox (git clone into a temp dir), confirming the +// verbatim-ported sandbox.HostSandbox behavior end-to-end: tool dispatch +// hits the real filesystem/shell, a commit made via run_bash is pushed back +// to the origin repo on a clean finish, and the sandbox temp dir is removed +// afterward — exactly the git-clone/push/cleanup sequence LocalRunner used +// to run inline. No prior test (LocalRunner's included) exercised this path; +// it's new coverage for logic that was previously only exercised manually. +func TestNativeRunner_Run_ProjectDir_SandboxToolsAndPush(t *testing.T) { + if _, err := exec.LookPath("git"); err != nil { + t.Skip("git not available") + } + + // Bare origin repo so `git push origin HEAD` from the sandbox clone + // doesn't hit receive.denyCurrentBranch. + origin := t.TempDir() + run := func(dir string, args ...string) { + t.Helper() + cmd := exec.Command("git", args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), + "GIT_AUTHOR_NAME=test", "GIT_AUTHOR_EMAIL=test@example.com", + "GIT_COMMITTER_NAME=test", "GIT_COMMITTER_EMAIL=test@example.com", + ) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v\n%s", args, err, out) + } + } + run(origin, "init", "--bare", "-b", "main") + + // Seed the bare repo with an initial commit (an empty bare repo has no + // HEAD, so LocalRunner's plain `git clone <remote> <dir>` needs something + // to check out). + seed := t.TempDir() + run(seed, "init", "-b", "main") + run(seed, "remote", "add", "origin", origin) + if err := os.WriteFile(filepath.Join(seed, "README.md"), []byte("seed\n"), 0o644); err != nil { + t.Fatal(err) + } + run(seed, "add", "-A") + run(seed, "commit", "-m", "seed") + run(seed, "push", "origin", "main") + + srv := fakeChatServer(t, []fakeTurn{ + {toolCalls: []llm.ToolCall{toolCall("c1", "write_file", `{"path":"hello.txt","content":"hello world"}`)}}, + {toolCalls: []llm.ToolCall{toolCall("c2", "read_file", `{"path":"hello.txt"}`)}}, + {toolCalls: []llm.ToolCall{toolCall("c3", "glob", `{"pattern":"*.txt"}`)}}, + {toolCalls: []llm.ToolCall{toolCall("c4", "run_bash", `{"command":"git add -A && git -c user.name=test -c user.email=test@example.com commit -m 'add hello'"}`)}}, + {content: "## Summary\nAdded hello.txt and committed."}, + }) + defer srv.Close() + + r := newLocalRunner(t, srv) + tt := localTask() + tt.Agent.ProjectDir = origin + exec_ := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} + + if err := r.Run(context.Background(), tt, exec_, newStoreChannel(&fakeChannelStore{}, tt.ID)); err != nil { + t.Fatalf("Run: %v", err) + } + + if exec_.SandboxDir != "" { + t.Errorf("expected SandboxDir cleared after clean finish, got %q", exec_.SandboxDir) + } + + // Verify the commit actually landed on origin. + logCmd := exec.Command("git", "-C", origin, "log", "--oneline", "main") + out, err := logCmd.CombinedOutput() + if err != nil { + t.Fatalf("git log on origin: %v\n%s", err, out) + } + if !strings.Contains(string(out), "add hello") { + t.Errorf("expected pushed commit 'add hello' in origin log, got:\n%s", out) + } +} |