diff options
Diffstat (limited to 'internal/executor')
| -rw-r--r-- | internal/executor/agentmcp.go | 160 | ||||
| -rw-r--r-- | internal/executor/agentmcp_test.go | 218 | ||||
| -rw-r--r-- | internal/executor/channel.go | 159 | ||||
| -rw-r--r-- | internal/executor/channel_test.go | 124 | ||||
| -rw-r--r-- | internal/executor/claude.go | 554 | ||||
| -rw-r--r-- | internal/executor/claude_test.go | 810 | ||||
| -rw-r--r-- | internal/executor/container.go | 147 | ||||
| -rw-r--r-- | internal/executor/container_test.go | 177 | ||||
| -rw-r--r-- | internal/executor/executor.go | 67 | ||||
| -rw-r--r-- | internal/executor/executor_test.go | 65 | ||||
| -rw-r--r-- | internal/executor/gemini.go | 346 | ||||
| -rw-r--r-- | internal/executor/gemini_test.go | 447 | ||||
| -rw-r--r-- | internal/executor/helpers.go | 26 | ||||
| -rw-r--r-- | internal/executor/local.go | 114 | ||||
| -rw-r--r-- | internal/executor/local_test.go | 197 | ||||
| -rw-r--r-- | internal/executor/localtools.go | 135 | ||||
| -rw-r--r-- | internal/executor/preamble.go | 67 | ||||
| -rw-r--r-- | internal/executor/preamble_test.go | 30 |
18 files changed, 1425 insertions, 2418 deletions
diff --git a/internal/executor/agentmcp.go b/internal/executor/agentmcp.go new file mode 100644 index 0000000..4368031 --- /dev/null +++ b/internal/executor/agentmcp.go @@ -0,0 +1,160 @@ +package executor + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "net/http" + "strings" + "sync" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// Registry maps per-task MCP bearer tokens to a built agent MCP server. A token +// is minted when a runner spawns an agent subprocess and revoked when it exits, +// so the server resolves task context from the token alone and never trusts an +// agent-supplied task ID. +type Registry struct { + mu sync.RWMutex + servers map[string]*mcp.Server +} + +func NewRegistry() *Registry { + return &Registry{servers: make(map[string]*mcp.Server)} +} + +// Mint creates a token bound to a freshly built MCP server for ch. +func (r *Registry) Mint(ch AgentChannel) (string, error) { + buf := make([]byte, 32) + if _, err := rand.Read(buf); err != nil { + return "", err + } + token := hex.EncodeToString(buf) + r.mu.Lock() + r.servers[token] = newAgentServer(ch) + r.mu.Unlock() + return token, nil +} + +func (r *Registry) server(token string) (*mcp.Server, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + s, ok := r.servers[token] + return s, ok +} + +func (r *Registry) Revoke(token string) { + r.mu.Lock() + delete(r.servers, token) + r.mu.Unlock() +} + +type askUserInput struct { + Question string `json:"question" jsonschema:"the question to ask the user; phrase it as a real question ending with a question mark"` + Options []string `json:"options,omitempty" jsonschema:"optional list of suggested answer choices"` +} + +type reportSummaryInput struct { + Summary string `json:"summary" jsonschema:"a 2-5 sentence summary of what you did and the outcome"` +} + +type spawnSubtaskInput struct { + Name string `json:"name" jsonschema:"short descriptive name for the subtask"` + Instructions string `json:"instructions" jsonschema:"complete instructions for the subtask agent"` + Model string `json:"model,omitempty" jsonschema:"optional model override, e.g. sonnet or opus"` + MaxBudgetUSD float64 `json:"max_budget_usd,omitempty" jsonschema:"optional budget cap in USD"` +} + +type recordProgressInput struct { + Message string `json:"message" jsonschema:"a short progress note describing what you are doing"` +} + +func textResult(text string) *mcp.CallToolResult { + return &mcp.CallToolResult{Content: []mcp.Content{&mcp.TextContent{Text: text}}} +} + +// newAgentServer builds an MCP server exposing the four agent tools bound to ch. +func newAgentServer(ch AgentChannel) *mcp.Server { + s := mcp.NewServer(&mcp.Implementation{Name: "claudomator", Version: "1"}, nil) + + mcp.AddTool(s, &mcp.Tool{ + Name: "ask_user", + Description: "Ask the user a question when you genuinely need a decision to proceed. Your turn ends after calling this; the task is resumed with the user's answer. Prefer making a reasonable decision and noting it in report_summary over asking.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in askUserInput) (*mcp.CallToolResult, any, error) { + q := map[string]any{"text": in.Question} + if len(in.Options) > 0 { + q["options"] = in.Options + } + payload, _ := json.Marshal(q) + ans, err := ch.AskUser(ctx, string(payload)) + if errors.Is(err, ErrAgentBlocked) { + return textResult("Question recorded. End your turn now without calling any more tools; the task will be resumed once the user answers."), nil, nil + } + if err != nil { + return nil, nil, err + } + return textResult(ans), nil, nil + }) + + mcp.AddTool(s, &mcp.Tool{ + Name: "report_summary", + Description: "Record a concise summary of what you accomplished. Call this before finishing.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in reportSummaryInput) (*mcp.CallToolResult, any, error) { + if err := ch.ReportSummary(ctx, in.Summary); err != nil { + return nil, nil, err + } + return textResult("Summary recorded."), nil, nil + }) + + mcp.AddTool(s, &mcp.Tool{ + Name: "spawn_subtask", + Description: "Create a child task to be executed separately. Use this to break large work into focused pieces, then finish your turn.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in spawnSubtaskInput) (*mcp.CallToolResult, any, error) { + id, err := ch.SpawnSubtask(ctx, SubtaskSpec{ + Name: in.Name, + Instructions: in.Instructions, + Model: in.Model, + MaxBudgetUSD: in.MaxBudgetUSD, + }) + if err != nil { + return nil, nil, err + } + return textResult("Created subtask " + id), nil, nil + }) + + mcp.AddTool(s, &mcp.Tool{ + Name: "record_progress", + Description: "Record a short progress note that appears in the task timeline.", + }, func(ctx context.Context, _ *mcp.CallToolRequest, in recordProgressInput) (*mcp.CallToolResult, any, error) { + if err := ch.RecordProgress(ctx, in.Message); err != nil { + return nil, nil, err + } + return textResult("Noted."), nil, nil + }) + + return s +} + +func bearerToken(r *http.Request) string { + h := r.Header.Get("Authorization") + if h == "" { + return "" + } + return strings.TrimSpace(strings.TrimPrefix(h, "Bearer ")) +} + +// NewAgentMCPHandler returns the HTTP handler for the per-task agent MCP server. +// It resolves the request's bearer token to the server for that task's run; +// unknown tokens yield a 400 from the underlying handler. +func NewAgentMCPHandler(reg *Registry) http.Handler { + return mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server { + s, ok := reg.server(bearerToken(r)) + if !ok { + return nil + } + return s + }, nil) +} diff --git a/internal/executor/agentmcp_test.go b/internal/executor/agentmcp_test.go new file mode 100644 index 0000000..3973af0 --- /dev/null +++ b/internal/executor/agentmcp_test.go @@ -0,0 +1,218 @@ +package executor + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +// recordingChannel is a fake AgentChannel that records tool invocations. +type recordingChannel struct { + asked string + summary string + spawned []SubtaskSpec + progress []string + spawnID string +} + +func (c *recordingChannel) AskUser(_ context.Context, q string) (string, error) { + c.asked = q + return "", ErrAgentBlocked +} +func (c *recordingChannel) ReportSummary(_ context.Context, s string) error { + c.summary = s + return nil +} +func (c *recordingChannel) SpawnSubtask(_ context.Context, spec SubtaskSpec) (string, error) { + c.spawned = append(c.spawned, spec) + return c.spawnID, nil +} +func (c *recordingChannel) RecordProgress(_ context.Context, m string) error { + c.progress = append(c.progress, m) + return nil +} + +func resultText(t *testing.T, res *mcp.CallToolResult) string { + t.Helper() + if len(res.Content) == 0 { + t.Fatal("expected content in tool result") + } + tc, ok := res.Content[0].(*mcp.TextContent) + if !ok { + t.Fatalf("expected TextContent, got %T", res.Content[0]) + } + return tc.Text +} + +func TestRegistry_MintLookupRevoke(t *testing.T) { + reg := NewRegistry() + tok, err := reg.Mint(&recordingChannel{}) + if err != nil { + t.Fatalf("Mint: %v", err) + } + if tok == "" { + t.Fatal("expected non-empty token") + } + if _, ok := reg.server(tok); !ok { + t.Error("expected server for minted token") + } + if _, ok := reg.server("nonsense"); ok { + t.Error("expected no server for unknown token") + } + reg.Revoke(tok) + if _, ok := reg.server(tok); ok { + t.Error("expected no server after revoke") + } +} + +func TestRegistry_MintUniqueTokens(t *testing.T) { + reg := NewRegistry() + a, _ := reg.Mint(&recordingChannel{}) + b, _ := reg.Mint(&recordingChannel{}) + if a == b { + t.Error("expected unique tokens per mint") + } +} + +// connectInMemory wires a client to a server over the SDK's in-memory transport. +func connectInMemory(t *testing.T, srv *mcp.Server) *mcp.ClientSession { + t.Helper() + ctx := context.Background() + clientT, serverT := mcp.NewInMemoryTransports() + if _, err := srv.Connect(ctx, serverT, nil); err != nil { + t.Fatalf("server connect: %v", err) + } + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + cs, err := client.Connect(ctx, clientT, nil) + if err != nil { + t.Fatalf("client connect: %v", err) + } + t.Cleanup(func() { _ = cs.Close() }) + return cs +} + +func TestAgentServer_AskUser_RecordsAndInstructsStop(t *testing.T) { + ch := &recordingChannel{} + cs := connectInMemory(t, newAgentServer(ch)) + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "ask_user", + Arguments: map[string]any{"question": "Proceed?", "options": []string{"yes", "no"}}, + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if !strings.Contains(ch.asked, `"text":"Proceed?"`) || !strings.Contains(ch.asked, "yes") { + t.Errorf("channel did not receive question+options: %q", ch.asked) + } + if txt := resultText(t, res); !strings.Contains(strings.ToLower(txt), "recorded") { + t.Errorf("expected stop instruction, got %q", txt) + } +} + +func TestAgentServer_ReportSummary(t *testing.T) { + ch := &recordingChannel{} + cs := connectInMemory(t, newAgentServer(ch)) + if _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "report_summary", + Arguments: map[string]any{"summary": "did the work"}, + }); err != nil { + t.Fatalf("CallTool: %v", err) + } + if ch.summary != "did the work" { + t.Errorf("summary not recorded, got %q", ch.summary) + } +} + +func TestAgentServer_SpawnSubtask(t *testing.T) { + ch := &recordingChannel{spawnID: "sub-1"} + cs := connectInMemory(t, newAgentServer(ch)) + res, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "spawn_subtask", + Arguments: map[string]any{"name": "child", "instructions": "do it", "model": "sonnet"}, + }) + if err != nil { + t.Fatalf("CallTool: %v", err) + } + if len(ch.spawned) != 1 || ch.spawned[0].Name != "child" || ch.spawned[0].Instructions != "do it" || ch.spawned[0].Model != "sonnet" { + t.Errorf("subtask spec not propagated: %+v", ch.spawned) + } + if txt := resultText(t, res); !strings.Contains(txt, "sub-1") { + t.Errorf("expected returned subtask ID in result, got %q", txt) + } +} + +func TestAgentServer_RecordProgress(t *testing.T) { + ch := &recordingChannel{} + cs := connectInMemory(t, newAgentServer(ch)) + if _, err := cs.CallTool(context.Background(), &mcp.CallToolParams{ + Name: "record_progress", + Arguments: map[string]any{"message": "halfway"}, + }); err != nil { + t.Fatalf("CallTool: %v", err) + } + if len(ch.progress) != 1 || ch.progress[0] != "halfway" { + t.Errorf("progress not recorded: %+v", ch.progress) + } +} + +type bearerRT struct { + token string + base http.RoundTripper +} + +func (b bearerRT) RoundTrip(r *http.Request) (*http.Response, error) { + r = r.Clone(r.Context()) + if b.token != "" { + r.Header.Set("Authorization", "Bearer "+b.token) + } + return b.base.RoundTrip(r) +} + +func TestAgentMCPHandler_HTTP_AuthAndDispatch(t *testing.T) { + ch := &recordingChannel{} + reg := NewRegistry() + tok, _ := reg.Mint(ch) + + httpSrv := httptest.NewServer(NewAgentMCPHandler(reg)) + defer httpSrv.Close() + + ctx := context.Background() + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + cs, err := client.Connect(ctx, &mcp.StreamableClientTransport{ + Endpoint: httpSrv.URL, + HTTPClient: &http.Client{Transport: bearerRT{token: tok, base: http.DefaultTransport}}, + }, nil) + if err != nil { + t.Fatalf("client connect with valid token: %v", err) + } + defer cs.Close() + + if _, err := cs.CallTool(ctx, &mcp.CallToolParams{ + Name: "record_progress", + Arguments: map[string]any{"message": "over http"}, + }); err != nil { + t.Fatalf("CallTool over HTTP: %v", err) + } + if len(ch.progress) != 1 || ch.progress[0] != "over http" { + t.Errorf("HTTP dispatch did not reach channel: %+v", ch.progress) + } +} + +func TestAgentMCPHandler_HTTP_BadTokenRejected(t *testing.T) { + reg := NewRegistry() + httpSrv := httptest.NewServer(NewAgentMCPHandler(reg)) + defer httpSrv.Close() + + client := mcp.NewClient(&mcp.Implementation{Name: "test", Version: "1"}, nil) + _, err := client.Connect(context.Background(), &mcp.StreamableClientTransport{ + Endpoint: httpSrv.URL, + HTTPClient: &http.Client{Transport: bearerRT{token: "invalid", base: http.DefaultTransport}}, + }, nil) + if err == nil { + t.Fatal("expected connect to fail with invalid token") + } +} diff --git a/internal/executor/channel.go b/internal/executor/channel.go new file mode 100644 index 0000000..8605ffa --- /dev/null +++ b/internal/executor/channel.go @@ -0,0 +1,159 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + "sync" + "time" + + "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 +} + +// channelStore is the subset of storage the default channel needs. +type channelStore interface { + CreateTask(t *task.Task) error + CreateEvent(e *event.Event) error +} + +// pendingAsker is implemented by channels that buffer an ask_user call so the +// runner can convert it into a BlockedError after the agent subprocess exits. +type pendingAsker interface { + PendingQuestion() (questionJSON string, blocked bool) +} + +// channelPendingQuestion reports whether the agent asked a question via the +// channel during the run (the MCP transport's ask_user path). +func channelPendingQuestion(ch AgentChannel) (string, bool) { + if pa, ok := ch.(pendingAsker); ok { + return pa.PendingQuestion() + } + return "", false +} + +// storeChannel is the default AgentChannel backed by storage. Summary and +// question signals are buffered here under a mutex (they may arrive on an MCP +// handler goroutine mid-run); the pool flushes the summary to the execution and +// the runner reads the pending question to build a BlockedError after the +// subprocess exits. SpawnSubtask/RecordProgress write immediately. +type storeChannel struct { + store channelStore + taskID string + + mu sync.Mutex + summary string + summarySet bool + question string + blocked bool +} + +var _ AgentChannel = (*storeChannel)(nil) + +func newStoreChannel(store channelStore, taskID string) *storeChannel { + return &storeChannel{store: store, taskID: taskID} +} + +// AskUser buffers the question and flags the run as blocked. The default +// transport cannot deliver an answer in-session, so it returns ErrAgentBlocked; +// the runner converts the buffered question into a BlockedError after exit, and +// question persistence is owned by the pool's BlockedError handling. +func (c *storeChannel) AskUser(_ context.Context, questionJSON string) (string, error) { + c.mu.Lock() + c.question = questionJSON + c.blocked = true + c.mu.Unlock() + return "", ErrAgentBlocked +} + +// ReportSummary buffers the summary; the pool flushes it onto the execution +// after the run (preferring it over its extract/synthesize fallbacks). +func (c *storeChannel) ReportSummary(_ context.Context, summary string) error { + c.mu.Lock() + c.summary = summary + c.summarySet = true + c.mu.Unlock() + return nil +} + +// ReportedSummary returns the buffered summary if report_summary was called. +func (c *storeChannel) ReportedSummary() (string, bool) { + c.mu.Lock() + defer c.mu.Unlock() + return c.summary, c.summarySet +} + +// PendingQuestion returns the buffered ask_user question if the run is blocked. +func (c *storeChannel) PendingQuestion() (string, bool) { + c.mu.Lock() + defer c.mu.Unlock() + return c.question, c.blocked +} + +func (c *storeChannel) SpawnSubtask(_ context.Context, spec SubtaskSpec) (string, error) { + now := time.Now().UTC() + child := &task.Task{ + ID: uuid.NewString(), + Name: spec.Name, + ParentTaskID: c.taskID, + Agent: task.AgentConfig{ + Type: "claude", + Model: spec.Model, + Instructions: spec.Instructions, + MaxBudgetUSD: spec.MaxBudgetUSD, + }, + Priority: task.PriorityNormal, + State: task.StatePending, + CreatedAt: now, + UpdatedAt: now, + } + if err := c.store.CreateTask(child); err != nil { + return "", err + } + return child.ID, nil +} + +func (c *storeChannel) RecordProgress(_ context.Context, message string) error { + payload, _ := json.Marshal(struct { + Message string `json:"message"` + }{Message: message}) + return c.store.CreateEvent(&event.Event{ + TaskID: c.taskID, + Kind: event.KindAgentMessage, + Actor: event.ActorAgent, + Payload: payload, + }) +}
\ No newline at end of file diff --git a/internal/executor/channel_test.go b/internal/executor/channel_test.go new file mode 100644 index 0000000..250e8eb --- /dev/null +++ b/internal/executor/channel_test.go @@ -0,0 +1,124 @@ +package executor + +import ( + "context" + "errors" + "testing" + + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/task" +) + +type fakeChannelStore struct { + createdTasks []*task.Task + createdEvents []*event.Event + createTaskErr error +} + +func (f *fakeChannelStore) CreateTask(t *task.Task) error { + if f.createTaskErr != nil { + return f.createTaskErr + } + f.createdTasks = append(f.createdTasks, t) + return nil +} + +func (f *fakeChannelStore) CreateEvent(e *event.Event) error { + f.createdEvents = append(f.createdEvents, e) + return nil +} + +func TestStoreChannel_AskUser_BuffersAndBlocks(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1") + answer, err := ch.AskUser(context.Background(), `{"text":"q"}`) + if !errors.Is(err, ErrAgentBlocked) { + t.Errorf("expected ErrAgentBlocked, got %v", err) + } + if answer != "" { + t.Errorf("expected empty answer, got %q", answer) + } + q, blocked := ch.PendingQuestion() + if !blocked || q != `{"text":"q"}` { + t.Errorf("expected buffered question, got q=%q blocked=%v", q, blocked) + } +} + +func TestStoreChannel_PendingQuestion_DefaultNotBlocked(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1") + if q, blocked := ch.PendingQuestion(); blocked || q != "" { + t.Errorf("expected no pending question, got q=%q blocked=%v", q, blocked) + } +} + +func TestStoreChannel_ReportSummary_Buffers(t *testing.T) { + ch := newStoreChannel(&fakeChannelStore{}, "task-1") + if _, ok := ch.ReportedSummary(); ok { + t.Error("expected no summary before report") + } + if err := ch.ReportSummary(context.Background(), "did the thing"); err != nil { + t.Fatalf("ReportSummary: %v", err) + } + sum, ok := ch.ReportedSummary() + if !ok || sum != "did the thing" { + t.Errorf("expected buffered summary, got %q ok=%v", sum, ok) + } +} + +func TestStoreChannel_SpawnSubtask_CreatesChildWithParent(t *testing.T) { + store := &fakeChannelStore{} + ch := newStoreChannel(store, "parent-1") + id, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{ + Name: "child", + Instructions: "do it", + Model: "sonnet", + MaxBudgetUSD: 1.5, + }) + if err != nil { + t.Fatalf("SpawnSubtask: %v", err) + } + if id == "" { + t.Error("expected non-empty subtask ID") + } + if len(store.createdTasks) != 1 { + t.Fatalf("expected 1 created task, got %d", len(store.createdTasks)) + } + ct := store.createdTasks[0] + if ct.ParentTaskID != "parent-1" { + t.Errorf("ParentTaskID: got %q want parent-1", ct.ParentTaskID) + } + if ct.ID != id { + t.Errorf("returned id %q != created task id %q", id, ct.ID) + } + if ct.Agent.Instructions != "do it" || ct.Agent.Model != "sonnet" || ct.Agent.MaxBudgetUSD != 1.5 { + t.Errorf("agent config not propagated: %+v", ct.Agent) + } + if ct.State != task.StatePending { + t.Errorf("expected PENDING child, got %s", ct.State) + } +} + +func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) { + store := &fakeChannelStore{createTaskErr: errors.New("boom")} + ch := newStoreChannel(store, "parent-1") + if _, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{Name: "x"}); err == nil { + t.Error("expected error from CreateTask") + } +} + +func TestStoreChannel_RecordProgress_EmitsAgentMessage(t *testing.T) { + store := &fakeChannelStore{} + ch := newStoreChannel(store, "task-1") + if err := ch.RecordProgress(context.Background(), "halfway done"); err != nil { + t.Fatalf("RecordProgress: %v", err) + } + if len(store.createdEvents) != 1 { + t.Fatalf("expected 1 event, got %d", len(store.createdEvents)) + } + e := store.createdEvents[0] + if e.Kind != event.KindAgentMessage || e.Actor != event.ActorAgent { + t.Errorf("got kind=%v actor=%v want agent_message/agent", e.Kind, e.Actor) + } + if e.TaskID != "task-1" { + t.Errorf("TaskID: got %q want task-1", e.TaskID) + } +} diff --git a/internal/executor/claude.go b/internal/executor/claude.go deleted file mode 100644 index 7b70486..0000000 --- a/internal/executor/claude.go +++ /dev/null @@ -1,554 +0,0 @@ -package executor - -import ( - "context" - "fmt" - "log/slog" - "os" - "os/exec" - "path/filepath" - "strings" - "sync" - "syscall" - "time" - - "github.com/thepeterstone/claudomator/internal/retry" - "github.com/thepeterstone/claudomator/internal/storage" - "github.com/thepeterstone/claudomator/internal/task" -) - -// ClaudeRunner spawns the `claude` CLI in non-interactive mode. -type ClaudeRunner struct { - BinaryPath string // defaults to "claude" - Logger *slog.Logger - LogDir string // base directory for execution logs - APIURL string // base URL of the Claudomator API, passed to subprocesses -} - -// 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. -// ExecLogDir returns the log directory for the given execution ID. -// Implements LogPather so the pool can persist paths before execution starts. -func (r *ClaudeRunner) ExecLogDir(execID string) string { - if r.LogDir == "" { - return "" - } - return filepath.Join(r.LogDir, execID) -} - -func (r *ClaudeRunner) binaryPath() string { - if r.BinaryPath != "" { - return r.BinaryPath - } - return "claude" -} - -// Run executes a claude -p invocation, streaming output to log files. -// It retries up to 3 times on rate-limit errors using exponential backoff. -// If the agent writes a question file and exits, Run returns *BlockedError. -// -// When project_dir is set and this is not a resume execution, Run clones the -// project into a temp sandbox, runs the agent there, then merges committed -// changes back to project_dir. On failure the sandbox is preserved and its -// path is included in the error. -func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { - projectDir := t.Agent.ProjectDir - - // Validate project_dir exists when set. - if projectDir != "" { - if _, err := os.Stat(projectDir); err != nil { - return fmt.Errorf("project_dir %q: %w", projectDir, err) - } - } - - // Setup log directory once; retries overwrite the log files. - logDir := r.ExecLogDir(e.ID) - if logDir == "" { - logDir = e.ID // fallback for tests without LogDir set - } - if err := os.MkdirAll(logDir, 0700); err != nil { - return fmt.Errorf("creating log dir: %w", err) - } - if e.StdoutPath == "" { - e.StdoutPath = filepath.Join(logDir, "stdout.log") - e.StderrPath = filepath.Join(logDir, "stderr.log") - e.ArtifactDir = logDir - } - - // Pre-assign session ID so we can resume after a BLOCKED state. - // For resume executions, the claude session continues under the original - // session ID (the one passed to --resume). Using the new exec's own UUID - // would cause a second block-and-resume cycle to pass the wrong --resume - // argument. - if e.SessionID == "" { - if e.ResumeSessionID != "" { - e.SessionID = e.ResumeSessionID - } else { - e.SessionID = e.ID // reuse execution UUID as session UUID (both are UUIDs) - } - } - - // For new (non-resume) executions with a project_dir, clone into a sandbox. - // Resume executions run in the preserved sandbox (e.SandboxDir) so Claude - // finds its session files under the same project slug. If no sandbox was - // preserved (e.g. task had no project_dir), fall back to project_dir. - var sandboxDir string - var startHEAD string - effectiveWorkingDir := projectDir - if e.ResumeSessionID != "" { - if e.SandboxDir != "" { - if _, statErr := os.Stat(e.SandboxDir); statErr == nil { - effectiveWorkingDir = e.SandboxDir - } else { - // Preserved sandbox was cleaned up (e.g. /tmp purge after reboot). - // Clone a fresh sandbox so the task can run rather than fail immediately. - r.Logger.Warn("preserved sandbox missing, cloning fresh", "sandbox", e.SandboxDir, "project_dir", projectDir) - e.SandboxDir = "" - if projectDir != "" { - var err error - sandboxDir, err = setupSandbox(t.Agent.ProjectDir, r.Logger) - if err != nil { - return fmt.Errorf("setting up sandbox: %w", err) - } - - effectiveWorkingDir = sandboxDir - r.Logger.Info("fresh sandbox created for resume", "sandbox", sandboxDir, "project_dir", projectDir) - } - } - } - } else if projectDir != "" { - var err error - sandboxDir, err = setupSandbox(t.Agent.ProjectDir, r.Logger) - if err != nil { - return fmt.Errorf("setting up sandbox: %w", err) - } - - effectiveWorkingDir = sandboxDir - r.Logger.Info("sandbox created", "sandbox", sandboxDir, "project_dir", projectDir) - } - - if effectiveWorkingDir != "" { - // Capture the initial HEAD so we can identify new commits later. - headOut, _ := exec.Command("git", gitSafe("-C", effectiveWorkingDir, "rev-parse", "HEAD")...).Output() - startHEAD = strings.TrimSpace(string(headOut)) - } - - questionFile := filepath.Join(logDir, "question.json") - args := r.buildArgs(t, e, questionFile) - - attempt := 0 - err := retry.RunWithBackoff(ctx, 3, 5*time.Second, func() error { - if attempt > 0 { - delay := 5 * time.Second * (1 << (attempt - 1)) - r.Logger.Warn("rate-limited by Claude API, retrying", - "attempt", attempt, - "delay", delay, - ) - } - attempt++ - return r.execOnce(ctx, args, effectiveWorkingDir, projectDir, e) - }) - if err != nil { - if sandboxDir != "" { - return fmt.Errorf("%w (sandbox preserved at %s)", err, sandboxDir) - } - return err - } - - // Check whether the agent left a question before exiting. - data, readErr := os.ReadFile(questionFile) - if readErr == nil { - os.Remove(questionFile) // consumed - questionJSON := strings.TrimSpace(string(data)) - // If the agent wrote a completion report instead of a real question, - // extract the text as the summary and fall through to normal completion. - if isCompletionReport(questionJSON) { - r.Logger.Info("treating question file as completion report", "taskID", e.TaskID) - e.Summary = extractQuestionText(questionJSON) - } else { - // Preserve sandbox on BLOCKED — agent may have partial work and its - // Claude session files are stored under the sandbox's project slug. - // The resume execution must run in the same directory. - return &BlockedError{QuestionJSON: questionJSON, SessionID: e.SessionID, SandboxDir: sandboxDir} - } - } - - // Read agent summary if written. - summaryFile := filepath.Join(logDir, "summary.txt") - if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil { - os.Remove(summaryFile) // consumed - e.Summary = strings.TrimSpace(string(summaryData)) - } - - // Merge sandbox back to project_dir and clean up. - if sandboxDir != "" { - if mergeErr := teardownSandbox(projectDir, sandboxDir, startHEAD, r.Logger, e); mergeErr != nil { - return fmt.Errorf("sandbox teardown: %w (sandbox preserved at %s)", mergeErr, sandboxDir) - } - } - return nil -} - -// sandboxCloneSource returns the URL to clone the sandbox from. It prefers a -// remote named "local" (a local bare repo that accepts pushes cleanly), then -// falls back to "origin", then to the working copy path itself. -func sandboxCloneSource(projectDir string) string { - for _, remote := range []string{"local", "origin"} { - out, err := exec.Command("git", gitSafe("-C", projectDir, "remote", "get-url", remote)...).Output() - if err == nil { - u := strings.TrimSpace(string(out)) - if u != "" && (strings.HasPrefix(u, "/") || strings.HasPrefix(u, "file://")) { - return u - } - } - } - return projectDir -} - -// setupSandbox prepares a temporary git clone of projectDir. -// If projectDir is not a git repo it is initialised with an initial commit first. -func setupSandbox(projectDir string, logger *slog.Logger) (string, error) { - // Ensure projectDir is a git repo; initialise if not. - if err := exec.Command("git", gitSafe("-C", projectDir, "rev-parse", "--git-dir")...).Run(); err != nil { - cmds := [][]string{ - gitSafe("-C", projectDir, "init"), - gitSafe("-C", projectDir, "add", "-A"), - gitSafe("-C", projectDir, "commit", "--allow-empty", "-m", "chore: initial commit"), - } - for _, args := range cmds { - if out, err := exec.Command("git", args...).CombinedOutput(); err != nil { //nolint:gosec - return "", fmt.Errorf("git init %s: %w\n%s", projectDir, err, out) - } - } - } - - src := sandboxCloneSource(projectDir) - - tempDir, err := os.MkdirTemp("", "claudomator-sandbox-*") - if err != nil { - return "", fmt.Errorf("creating sandbox dir: %w", err) - } - // git clone requires the target to not exist; remove the placeholder first. - if err := os.Remove(tempDir); err != nil { - return "", fmt.Errorf("removing temp dir placeholder: %w", err) - } - out, err := exec.Command("git", gitSafe("clone", "--no-hardlinks", src, tempDir)...).CombinedOutput() - if err != nil { - return "", fmt.Errorf("git clone: %w\n%s", err, out) - } - return tempDir, nil -} - -// teardownSandbox verifies the sandbox is clean and pushes new commits to the -// canonical bare repo. If the push is rejected because another task pushed -// concurrently, it fetches and rebases then retries once. -// -// The working copy (projectDir) is NOT updated automatically — it is the -// developer's workspace and is pulled manually. This avoids permission errors -// from mixed-owner .git/objects directories. -func teardownSandbox(projectDir, sandboxDir, startHEAD string, logger *slog.Logger, execRecord *storage.Execution) error { - // Automatically commit uncommitted changes. - out, err := exec.Command("git", gitSafe("-C", sandboxDir, "status", "--porcelain")...).Output() - if err != nil { - return fmt.Errorf("git status: %w", err) - } - if len(strings.TrimSpace(string(out))) > 0 { - logger.Info("autocommitting uncommitted changes", "sandbox", sandboxDir) - - // Run build before autocommitting. - if _, err := os.Stat(filepath.Join(sandboxDir, "Makefile")); err == nil { - logger.Info("running 'make build' before autocommit", "sandbox", sandboxDir) - if buildOut, buildErr := exec.Command("make", "-C", sandboxDir, "build").CombinedOutput(); buildErr != nil { - return fmt.Errorf("build failed before autocommit: %w\n%s", buildErr, buildOut) - } - } else if _, err := os.Stat(filepath.Join(sandboxDir, "gradlew")); err == nil { - logger.Info("running './gradlew build' before autocommit", "sandbox", sandboxDir) - cmd := exec.Command("./gradlew", "build") - cmd.Dir = sandboxDir - if buildOut, buildErr := cmd.CombinedOutput(); buildErr != nil { - return fmt.Errorf("build failed before autocommit: %w\n%s", buildErr, buildOut) - } - } else if _, err := os.Stat(filepath.Join(sandboxDir, "go.mod")); err == nil { - logger.Info("running 'go build ./...' before autocommit", "sandbox", sandboxDir) - cmd := exec.Command("go", "build", "./...") - cmd.Dir = sandboxDir - if buildOut, buildErr := cmd.CombinedOutput(); buildErr != nil { - return fmt.Errorf("build failed before autocommit: %w\n%s", buildErr, buildOut) - } - } - - cmds := [][]string{ - gitSafe("-C", sandboxDir, "add", "-A"), - gitSafe("-C", sandboxDir, "commit", "-m", "chore: autocommit uncommitted changes"), - } - for _, args := range cmds { - if out, err := exec.Command("git", args...).CombinedOutput(); err != nil { - return fmt.Errorf("autocommit failed (%v): %w\n%s", args, err, out) - } - } - } - - // Capture commits before pushing/deleting. - // Use startHEAD..HEAD to find all commits made during this execution. - logRange := "origin/HEAD..HEAD" - if startHEAD != "" && startHEAD != "HEAD" { - logRange = startHEAD + "..HEAD" - } - - logCmd := exec.Command("git", gitSafe("-C", sandboxDir, "log", logRange, "--pretty=format:%H|%s")...) - logOut, logErr := logCmd.CombinedOutput() - if logErr == nil { - lines := strings.Split(strings.TrimSpace(string(logOut)), "\n") - logger.Debug("captured commits", "count", len(lines), "range", logRange) - for _, line := range lines { - if line == "" { - continue - } - parts := strings.SplitN(line, "|", 2) - if len(parts) == 2 { - execRecord.Commits = append(execRecord.Commits, task.GitCommit{ - Hash: parts[0], - Message: parts[1], - }) - } - } - } else { - logger.Warn("failed to capture commits", "err", logErr, "range", logRange, "output", string(logOut)) - } - - // Check whether there are any new commits to push. - ahead, err := exec.Command("git", gitSafe("-C", sandboxDir, "rev-list", "--count", logRange)...).Output() - if err != nil { - logger.Warn("could not determine commits ahead of origin; proceeding", "err", err, "range", logRange) - } - if strings.TrimSpace(string(ahead)) == "0" { - os.RemoveAll(sandboxDir) - return nil - } - - // Push from sandbox → bare repo (sandbox's origin is the bare repo). - if out, err := exec.Command("git", gitSafe("-C", sandboxDir, "push", "origin", "HEAD")...).CombinedOutput(); err != nil { - // If rejected due to concurrent push, fetch+rebase and retry once. - // Use `git pull --rebase` without explicit remote/branch so git uses the - // configured upstream (set by clone), handling both main- and master-based repos. - if strings.Contains(string(out), "fetch first") || strings.Contains(string(out), "non-fast-forward") { - logger.Info("push rejected (concurrent task); rebasing and retrying", "sandbox", sandboxDir) - if out2, err2 := exec.Command("git", gitSafe("-C", sandboxDir, "pull", "--rebase")...).CombinedOutput(); err2 != nil { - return fmt.Errorf("git rebase before retry push: %w\n%s", err2, out2) - } - // Re-capture commits after rebase (hashes might have changed) - execRecord.Commits = nil - logOut, logErr = exec.Command("git", gitSafe("-C", sandboxDir, "log", logRange, "--pretty=format:%H|%s")...).Output() - if logErr == nil { - lines := strings.Split(strings.TrimSpace(string(logOut)), "\n") - for _, line := range lines { - parts := strings.SplitN(line, "|", 2) - if len(parts) == 2 { - execRecord.Commits = append(execRecord.Commits, task.GitCommit{ - Hash: parts[0], - Message: parts[1], - }) - } - } - } - - if out3, err3 := exec.Command("git", gitSafe("-C", sandboxDir, "push", "origin", "HEAD")...).CombinedOutput(); err3 != nil { - return fmt.Errorf("git push to origin (after rebase): %w\n%s", err3, out3) - } - } else { - return fmt.Errorf("git push to origin: %w\n%s", err, out) - } - } - - logger.Info("sandbox pushed to bare repo", "sandbox", sandboxDir) - os.RemoveAll(sandboxDir) - return nil -} - -// execOnce runs the claude subprocess once, streaming output to e's log paths. -func (r *ClaudeRunner) execOnce(ctx context.Context, args []string, workingDir, projectDir string, e *storage.Execution) error { - cmd := exec.CommandContext(ctx, r.binaryPath(), args...) - cmd.Env = append(os.Environ(), - "CLAUDOMATOR_API_URL="+r.APIURL, - "CLAUDOMATOR_TASK_ID="+e.TaskID, - "CLAUDOMATOR_PROJECT_DIR="+projectDir, - "CLAUDOMATOR_QUESTION_FILE="+filepath.Join(e.ArtifactDir, "question.json"), - "CLAUDOMATOR_SUMMARY_FILE="+filepath.Join(e.ArtifactDir, "summary.txt"), - ) - // Put the subprocess in its own process group so we can SIGKILL the entire - // group (MCP servers, bash children, etc.) on cancellation. - cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} - if workingDir != "" { - cmd.Dir = workingDir - } - - stdoutFile, err := os.Create(e.StdoutPath) - if err != nil { - return fmt.Errorf("creating stdout log: %w", err) - } - defer stdoutFile.Close() - - stderrFile, err := os.Create(e.StderrPath) - if err != nil { - return fmt.Errorf("creating stderr log: %w", err) - } - defer stderrFile.Close() - - // Use os.Pipe for stdout so we own the read-end lifetime. - // cmd.StdoutPipe() would add the read-end to closeAfterWait, causing - // cmd.Wait() to close it before our goroutine finishes reading. - stdoutR, stdoutW, err := os.Pipe() - if err != nil { - return fmt.Errorf("creating stdout pipe: %w", err) - } - cmd.Stdout = stdoutW // *os.File — not added to closeAfterStart/Wait - cmd.Stderr = stderrFile - - if err := cmd.Start(); err != nil { - stdoutW.Close() - stdoutR.Close() - return fmt.Errorf("starting claude: %w", err) - } - // Close our write-end immediately; the subprocess holds its own copy. - // The goroutine below gets EOF when the subprocess exits. - stdoutW.Close() - - // killDone is closed when cmd.Wait() returns, stopping the pgid-kill goroutine. - // - // Safety: this goroutine cannot block indefinitely. The select has two arms: - // • ctx.Done() — fires if the caller cancels (e.g. timeout, user cancel). - // The goroutine sends SIGKILL and exits immediately. - // • killDone — closed by close(killDone) below, immediately after cmd.Wait() - // returns. This fires when the process exits for any reason (natural exit, - // SIGKILL from the ctx arm, or any other signal). The goroutine exits without - // doing anything. - // - // Therefore: for a task that completes normally with a long-lived (non-cancelled) - // context, the killDone arm fires and the goroutine exits. There is no path where - // this goroutine outlives execOnce(). - killDone := make(chan struct{}) - go func() { - select { - case <-ctx.Done(): - // SIGKILL the entire process group to reap orphan children. - syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) - case <-killDone: - } - }() - - // Stream stdout to the log file and parse cost/errors. - // wg ensures costUSD and streamErr are fully written before we read them after cmd.Wait(). - var costUSD float64 - var streamErr error - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - costUSD, _, streamErr = parseStream(stdoutR, stdoutFile, r.Logger) - stdoutR.Close() - }() - - waitErr := cmd.Wait() - close(killDone) // stop the pgid-kill goroutine - wg.Wait() // drain remaining stdout before reading costUSD/streamErr - - e.CostUSD = costUSD - - if waitErr != nil { - if exitErr, ok := waitErr.(*exec.ExitError); ok { - e.ExitCode = exitErr.ExitCode() - } - // If the stream captured a rate-limit or quota message, return it - // so callers can distinguish it from a generic exit-status failure. - if retry.IsRateLimitError(streamErr) || isQuotaExhausted(streamErr) { - return streamErr - } - if tail := tailFile(e.StderrPath, 20); tail != "" { - return fmt.Errorf("claude exited with error: %w\nstderr:\n%s", waitErr, tail) - } - return fmt.Errorf("claude exited with error: %w", waitErr) - } - - e.ExitCode = 0 - if streamErr != nil { - return streamErr - } - return nil -} - -func (r *ClaudeRunner) buildArgs(t *task.Task, e *storage.Execution, questionFile string) []string { - // Resume execution: the agent already has context; just deliver the answer. - if e.ResumeSessionID != "" { - args := []string{ - "-p", e.ResumeAnswer, - "--resume", e.ResumeSessionID, - "--output-format", "stream-json", - "--verbose", - } - permMode := t.Agent.PermissionMode - if permMode == "" { - permMode = "bypassPermissions" - } - args = append(args, "--permission-mode", permMode) - if t.Agent.Model != "" { - args = append(args, "--model", t.Agent.Model) - } - return args - } - - instructions := t.Agent.Instructions - allowedTools := t.Agent.AllowedTools - - if !t.Agent.SkipPlanning { - instructions = withPlanningPreamble(instructions) - // Ensure Bash is available so the agent can POST subtasks and ask questions. - hasBash := false - for _, tool := range allowedTools { - if tool == "Bash" { - hasBash = true - break - } - } - if !hasBash { - allowedTools = append(allowedTools, "Bash") - } - } - - args := []string{ - "-p", instructions, - "--session-id", e.SessionID, - "--output-format", "stream-json", - "--verbose", - } - - if t.Agent.Model != "" { - args = append(args, "--model", t.Agent.Model) - } - if t.Agent.MaxBudgetUSD > 0 { - args = append(args, "--max-budget-usd", fmt.Sprintf("%.2f", t.Agent.MaxBudgetUSD)) - } - // Default to bypassPermissions — claudomator runs tasks unattended, so - // prompting for write access would always stall execution. Tasks that need - // a more restrictive mode can set permission_mode explicitly. - permMode := t.Agent.PermissionMode - if permMode == "" { - permMode = "bypassPermissions" - } - args = append(args, "--permission-mode", permMode) - if t.Agent.SystemPromptAppend != "" { - args = append(args, "--append-system-prompt", t.Agent.SystemPromptAppend) - } - for _, tool := range allowedTools { - args = append(args, "--allowedTools", tool) - } - for _, tool := range t.Agent.DisallowedTools { - args = append(args, "--disallowedTools", tool) - } - for _, f := range t.Agent.ContextFiles { - args = append(args, "--add-dir", f) - } - args = append(args, t.Agent.AdditionalArgs...) - - return args -} - diff --git a/internal/executor/claude_test.go b/internal/executor/claude_test.go deleted file mode 100644 index c01e160..0000000 --- a/internal/executor/claude_test.go +++ /dev/null @@ -1,810 +0,0 @@ -package executor - -import ( - "context" - "fmt" - "io" - "log/slog" - "os" - "os/exec" - "path/filepath" - "runtime" - "strings" - "testing" - "time" - - "github.com/thepeterstone/claudomator/internal/storage" - "github.com/thepeterstone/claudomator/internal/task" -) - -func TestClaudeRunner_BuildArgs_BasicTask(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "fix the bug", - Model: "sonnet", - SkipPlanning: true, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - argMap := make(map[string]bool) - for _, a := range args { - argMap[a] = true - } - for _, want := range []string{"-p", "fix the bug", "--output-format", "stream-json", "--verbose", "--model", "sonnet"} { - if !argMap[want] { - t.Errorf("missing arg %q in %v", want, args) - } - } -} - -func TestClaudeRunner_BuildArgs_FullConfig(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "implement feature", - Model: "opus", - MaxBudgetUSD: 5.0, - PermissionMode: "bypassPermissions", - SystemPromptAppend: "Follow TDD", - AllowedTools: []string{"Bash", "Edit"}, - DisallowedTools: []string{"Write"}, - ContextFiles: []string{"/src"}, - AdditionalArgs: []string{"--verbose"}, - SkipPlanning: true, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - // Check key args are present. - argMap := make(map[string]bool) - for _, a := range args { - argMap[a] = true - } - - requiredArgs := []string{ - "-p", "implement feature", "--output-format", "stream-json", - "--model", "opus", "--max-budget-usd", "5.00", - "--permission-mode", "bypassPermissions", - "--append-system-prompt", "Follow TDD", - "--allowedTools", "Bash", "Edit", - "--disallowedTools", "Write", - "--add-dir", "/src", - "--verbose", - } - for _, req := range requiredArgs { - if !argMap[req] { - t.Errorf("missing arg %q in %v", req, args) - } - } -} - -func TestClaudeRunner_BuildArgs_DefaultsToBypassPermissions(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "do work", - SkipPlanning: true, - // PermissionMode intentionally not set - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - found := false - for i, a := range args { - if a == "--permission-mode" && i+1 < len(args) && args[i+1] == "bypassPermissions" { - found = true - } - } - if !found { - t.Errorf("expected --permission-mode bypassPermissions when PermissionMode is empty, args: %v", args) - } -} - -func TestClaudeRunner_BuildArgs_RespectsExplicitPermissionMode(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "do work", - PermissionMode: "default", - SkipPlanning: true, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - for i, a := range args { - if a == "--permission-mode" && i+1 < len(args) { - if args[i+1] != "default" { - t.Errorf("expected --permission-mode default, got %q", args[i+1]) - } - return - } - } - t.Errorf("--permission-mode flag not found in args: %v", args) -} - -func TestClaudeRunner_BuildArgs_AlwaysIncludesVerbose(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "do something", - SkipPlanning: true, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - found := false - for _, a := range args { - if a == "--verbose" { - found = true - break - } - } - if !found { - t.Errorf("--verbose missing from args: %v", args) - } -} - -func TestClaudeRunner_BuildArgs_PreamblePrepended(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "fix the bug", - SkipPlanning: false, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - // The -p value should start with the preamble and end with the original instructions. - if len(args) < 2 || args[0] != "-p" { - t.Fatalf("expected -p as first arg, got: %v", args) - } - if !strings.HasPrefix(args[1], "## Runtime Environment") { - t.Errorf("instructions should start with planning preamble, got prefix: %q", args[1][:min(len(args[1]), 20)]) - } - if !strings.Contains(args[1], "$CLAUDOMATOR_PROJECT_DIR") { - t.Errorf("preamble should mention $CLAUDOMATOR_PROJECT_DIR") - } - if !strings.HasSuffix(args[1], "fix the bug") { - t.Errorf("instructions should end with original instructions") - } -} - -func TestClaudeRunner_BuildArgs_PreambleAddsBash(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "do work", - AllowedTools: []string{"Read"}, - SkipPlanning: false, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - // Bash should be appended to allowed tools. - foundBash := false - for i, a := range args { - if a == "--allowedTools" && i+1 < len(args) && args[i+1] == "Bash" { - foundBash = true - } - } - if !foundBash { - t.Errorf("Bash should be added to --allowedTools when preamble is active: %v", args) - } -} - -func TestClaudeRunner_BuildArgs_PreambleBashNotDuplicated(t *testing.T) { - r := &ClaudeRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "do work", - AllowedTools: []string{"Bash", "Read"}, - SkipPlanning: false, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - // Count Bash occurrences in --allowedTools values. - bashCount := 0 - for i, a := range args { - if a == "--allowedTools" && i+1 < len(args) && args[i+1] == "Bash" { - bashCount++ - } - } - if bashCount != 1 { - t.Errorf("Bash should appear exactly once in --allowedTools, got %d: %v", bashCount, args) - } -} - -// TestClaudeRunner_Run_ResumeSetsSessionIDFromResumeSession verifies that when a -// resume execution is itself blocked again, the stored SessionID is the original -// resumed session, not the new execution's own UUID. Without this, a second -// block-and-resume cycle passes the wrong --resume session ID and fails. -func TestClaudeRunner_Run_ResumeSetsSessionIDFromResumeSession(t *testing.T) { - logDir := t.TempDir() - r := &ClaudeRunner{ - BinaryPath: "true", // exits 0, no output - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - Instructions: "continue", - SkipPlanning: true, - }, - } - exec := &storage.Execution{ - ID: "resume-exec-uuid", - TaskID: "task-1", - ResumeSessionID: "original-session-uuid", - ResumeAnswer: "yes", - } - - // Run completes successfully (binary is "true"). - _ = r.Run(context.Background(), tk, exec) - - // SessionID must be the original session (ResumeSessionID), not the new - // exec's own ID. If it were exec.ID, a second blocked-then-resumed cycle - // would use the wrong --resume argument and fail. - if exec.SessionID != "original-session-uuid" { - t.Errorf("SessionID after resume Run: want %q, got %q", "original-session-uuid", exec.SessionID) - } -} - -func TestClaudeRunner_Run_InaccessibleWorkingDir_ReturnsError(t *testing.T) { - r := &ClaudeRunner{ - BinaryPath: "true", // would succeed if it ran - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: t.TempDir(), - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - ProjectDir: "/nonexistent/path/does/not/exist", - SkipPlanning: true, - }, - } - exec := &storage.Execution{ID: "test-exec"} - - err := r.Run(context.Background(), tk, exec) - - if err == nil { - t.Fatal("expected error for inaccessible working_dir, got nil") - } - if !strings.Contains(err.Error(), "project_dir") { - t.Errorf("expected 'project_dir' in error, got: %v", err) - } -} - -func TestClaudeRunner_BinaryPath_Default(t *testing.T) { - r := &ClaudeRunner{} - if r.binaryPath() != "claude" { - t.Errorf("want 'claude', got %q", r.binaryPath()) - } -} - -func TestClaudeRunner_BinaryPath_Custom(t *testing.T) { - r := &ClaudeRunner{BinaryPath: "/usr/local/bin/claude"} - if r.binaryPath() != "/usr/local/bin/claude" { - t.Errorf("want custom path, got %q", r.binaryPath()) - } -} - -// TestExecOnce_NoGoroutineLeak_OnNaturalExit verifies that execOnce does not -// leave behind any goroutines when the subprocess exits normally (no context -// cancellation). Both the pgid-kill goroutine and the parseStream goroutine -// must have exited before execOnce returns. -func TestExecOnce_NoGoroutineLeak_OnNaturalExit(t *testing.T) { - logDir := t.TempDir() - r := &ClaudeRunner{ - BinaryPath: "true", // exits immediately with status 0, produces no output - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - e := &storage.Execution{ - ID: "goroutine-leak-test", - TaskID: "task-id", - StdoutPath: filepath.Join(logDir, "stdout.log"), - StderrPath: filepath.Join(logDir, "stderr.log"), - ArtifactDir: logDir, - } - - // Let any goroutines from test infrastructure settle before sampling. - runtime.Gosched() - baseline := runtime.NumGoroutine() - - if err := r.execOnce(context.Background(), []string{}, "", "", e); err != nil { - t.Fatalf("execOnce failed: %v", err) - } - - // Give the scheduler a moment to let any leaked goroutines actually exit. - // In correct code the goroutines exit before execOnce returns, so this is - // just a safety buffer for the scheduler. - time.Sleep(10 * time.Millisecond) - runtime.Gosched() - - after := runtime.NumGoroutine() - if after > baseline { - t.Errorf("goroutine leak: %d goroutines before execOnce, %d after (leaked %d)", - baseline, after, after-baseline) - } -} - -// initGitRepo creates a git repo in dir with one commit so it is clonable. -func initGitRepo(t *testing.T, dir string) { - t.Helper() - cmds := [][]string{ - {"git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", dir, "init", "-b", "main"}, - {"git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", dir, "config", "user.email", "test@test"}, - {"git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", dir, "config", "user.name", "test"}, - } - for _, args := range cmds { - if out, err := exec.Command(args[0], args[1:]...).CombinedOutput(); err != nil { - t.Fatalf("%v: %v\n%s", args, err, out) - } - } - if err := os.WriteFile(filepath.Join(dir, "init.txt"), []byte("init"), 0644); err != nil { - t.Fatal(err) - } - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", dir, "add", ".").CombinedOutput(); err != nil { - t.Fatalf("git add: %v\n%s", err, out) - } - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", dir, "commit", "-m", "init").CombinedOutput(); err != nil { - t.Fatalf("git commit: %v\n%s", err, out) - } -} - -func TestSandboxCloneSource_PrefersLocalRemote(t *testing.T) { - dir := t.TempDir() - initGitRepo(t, dir) - // Add a "local" remote pointing to a bare repo. - bare := t.TempDir() - exec.Command("git", "init", "--bare", bare).Run() - exec.Command("git", "-C", dir, "remote", "add", "local", bare).Run() - exec.Command("git", "-C", dir, "remote", "add", "origin", "https://example.com/repo").Run() - - got := sandboxCloneSource(dir) - if got != bare { - t.Errorf("expected bare repo path %q, got %q", bare, got) - } -} - -func TestSandboxCloneSource_FallsBackToOrigin(t *testing.T) { - dir := t.TempDir() - initGitRepo(t, dir) - // sandboxCloneSource intentionally filters to local-FS remotes (so - // `git clone <src>` doesn't go over the network). Use a local path - // for origin to verify the fallback semantics. - originURL := t.TempDir() - exec.Command("git", "-C", dir, "remote", "add", "origin", originURL).Run() - - got := sandboxCloneSource(dir) - if got != originURL { - t.Errorf("expected origin URL %q, got %q", originURL, got) - } -} - -func TestSandboxCloneSource_FallsBackToProjectDir(t *testing.T) { - dir := t.TempDir() - initGitRepo(t, dir) - // No remotes configured. - got := sandboxCloneSource(dir) - if got != dir { - t.Errorf("expected projectDir %q (no remotes), got %q", dir, got) - } -} - -func TestSetupSandbox_ClonesGitRepo(t *testing.T) { - src := t.TempDir() - initGitRepo(t, src) - - sandbox, err := setupSandbox(src, slog.Default()) - if err != nil { - t.Fatalf("setupSandbox: %v", err) - } - t.Cleanup(func() { os.RemoveAll(sandbox) }) - - // Force sandbox to master if it cloned as main - exec.Command("git", gitSafe("-C", sandbox, "checkout", "master")...).Run() - - // Debug sandbox - logOut, _ := exec.Command("git", "-C", sandbox, "log", "-1").CombinedOutput() - fmt.Printf("DEBUG: sandbox log: %s\n", string(logOut)) - - // Verify sandbox is a git repo with at least one commit. - out, err := exec.Command("git", "-C", sandbox, "log", "--oneline").Output() - if err != nil { - t.Fatalf("git log in sandbox: %v", err) - } - if len(strings.TrimSpace(string(out))) == 0 { - t.Error("expected at least one commit in sandbox, got empty log") - } -} - -func TestSetupSandbox_InitialisesNonGitDir(t *testing.T) { - // A plain directory (not a git repo) should be initialised then cloned. - src := t.TempDir() - - sandbox, err := setupSandbox(src, slog.Default()) - if err != nil { - t.Fatalf("setupSandbox on plain dir: %v", err) - } - t.Cleanup(func() { os.RemoveAll(sandbox) }) - - if _, err := os.Stat(filepath.Join(sandbox, ".git")); err != nil { - t.Errorf("sandbox should be a git repo: %v", err) - } -} - -func TestTeardownSandbox_AutocommitsChanges(t *testing.T) { - // Create a bare repo as origin so push succeeds. - bare := t.TempDir() - if out, err := exec.Command("git", "init", "--bare", "-b", "main", bare).CombinedOutput(); err != nil { - t.Fatalf("git init bare: %v\n%s", err, out) - } - - // Create a sandbox directly. - sandbox := t.TempDir() - initGitRepo(t, sandbox) - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "remote", "add", "origin", bare).CombinedOutput(); err != nil { - t.Fatalf("git remote add: %v\n%s", err, out) - } - // Initial push to establish origin/main - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "push", "origin", "main").CombinedOutput(); err != nil { - t.Fatalf("git push initial: %v\n%s", err, out) - } - - // Capture startHEAD - headOut, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "rev-parse", "HEAD").Output() - if err != nil { - t.Fatalf("rev-parse HEAD: %v", err) - } - startHEAD := strings.TrimSpace(string(headOut)) - - // Leave an uncommitted file in the sandbox. - if err := os.WriteFile(filepath.Join(sandbox, "dirty.txt"), []byte("autocommit me"), 0644); err != nil { - t.Fatal(err) - } - - logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) - execRecord := &storage.Execution{} - - err = teardownSandbox("", sandbox, startHEAD, logger, execRecord) - if err != nil { - t.Fatalf("expected autocommit to succeed, got error: %v", err) - } - - // Sandbox should be removed after successful autocommit and push. - if _, statErr := os.Stat(sandbox); !os.IsNotExist(statErr) { - t.Error("sandbox should have been removed after successful autocommit and push") - } - - // Verify the commit exists in the bare repo. - out, err := exec.Command("git", "-C", bare, "log", "-1", "--pretty=%B").Output() - if err != nil { - t.Fatalf("git log in bare repo: %v", err) - } - if !strings.Contains(string(out), "chore: autocommit uncommitted changes") { - t.Errorf("expected autocommit message in log, got: %q", string(out)) - } - - // Verify the commit was captured in execRecord. - if len(execRecord.Commits) == 0 { - t.Error("expected at least one commit in execRecord") - } else if !strings.Contains(execRecord.Commits[0].Message, "chore: autocommit uncommitted changes") { - t.Errorf("unexpected commit message: %q", execRecord.Commits[0].Message) - } -} - -func TestTeardownSandbox_BuildFailure_BlocksAutocommit(t *testing.T) { - bare := t.TempDir() - if out, err := exec.Command("git", "init", "--bare", "-b", "main", bare).CombinedOutput(); err != nil { - t.Fatalf("git init bare: %v\n%s", err, out) - } - - sandbox := t.TempDir() - initGitRepo(t, sandbox) - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "remote", "add", "origin", bare).CombinedOutput(); err != nil { - t.Fatalf("git remote add: %v\n%s", err, out) - } - - // Capture startHEAD - headOut, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "rev-parse", "HEAD").Output() - if err != nil { - t.Fatalf("rev-parse HEAD: %v", err) - } - startHEAD := strings.TrimSpace(string(headOut)) - - // Leave an uncommitted file. - if err := os.WriteFile(filepath.Join(sandbox, "dirty.txt"), []byte("dirty"), 0644); err != nil { - t.Fatal(err) - } - - // Add a failing Makefile. - makefile := "build:\n\t@echo 'build failed'\n\texit 1\n" - if err := os.WriteFile(filepath.Join(sandbox, "Makefile"), []byte(makefile), 0644); err != nil { - t.Fatal(err) - } - - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - execRecord := &storage.Execution{} - - err = teardownSandbox("", sandbox, startHEAD, logger, execRecord) - if err == nil { - t.Error("expected teardown to fail due to build failure, but it succeeded") - } else if !strings.Contains(err.Error(), "build failed before autocommit") { - t.Errorf("expected build failure error message, got: %v", err) - } - - // Sandbox should NOT be removed if teardown failed. - if _, statErr := os.Stat(sandbox); os.IsNotExist(statErr) { - t.Error("sandbox should have been preserved after build failure") - } - - // Verify no new commit in bare repo. - out, err := exec.Command("git", "-C", bare, "log", "HEAD").CombinedOutput() - if strings.Contains(string(out), "chore: autocommit uncommitted changes") { - t.Error("autocommit should not have been pushed after build failure") - } -} - -func TestTeardownSandbox_BuildSuccess_ProceedsToAutocommit(t *testing.T) { - bare := t.TempDir() - if out, err := exec.Command("git", "init", "--bare", "-b", "main", bare).CombinedOutput(); err != nil { - t.Fatalf("git init bare: %v\n%s", err, out) - } - - sandbox := t.TempDir() - initGitRepo(t, sandbox) - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "remote", "add", "origin", bare).CombinedOutput(); err != nil { - t.Fatalf("git remote add: %v\n%s", err, out) - } - - // Capture startHEAD - headOut, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "rev-parse", "HEAD").Output() - if err != nil { - t.Fatalf("rev-parse HEAD: %v", err) - } - startHEAD := strings.TrimSpace(string(headOut)) - - // Leave an uncommitted file. - if err := os.WriteFile(filepath.Join(sandbox, "dirty.txt"), []byte("dirty"), 0644); err != nil { - t.Fatal(err) - } - - // Add a successful Makefile. - makefile := "build:\n\t@echo 'build succeeded'\n" - if err := os.WriteFile(filepath.Join(sandbox, "Makefile"), []byte(makefile), 0644); err != nil { - t.Fatal(err) - } - - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - execRecord := &storage.Execution{} - - err = teardownSandbox("", sandbox, startHEAD, logger, execRecord) - if err != nil { - t.Fatalf("expected teardown to succeed after build success, got error: %v", err) - } - - // Sandbox should be removed after success. - if _, statErr := os.Stat(sandbox); !os.IsNotExist(statErr) { - t.Error("sandbox should have been removed after successful build and autocommit") - } - - // Verify new commit in bare repo. - out, err := exec.Command("git", "-C", bare, "log", "-1", "--pretty=%B").Output() - if err != nil { - t.Fatalf("git log in bare repo: %v", err) - } - if !strings.Contains(string(out), "chore: autocommit uncommitted changes") { - t.Errorf("expected autocommit message in log, got: %q", string(out)) - } -} - - -func TestTeardownSandbox_CapturesExplicitCommits(t *testing.T) { - bare := t.TempDir() - if out, err := exec.Command("git", "init", "--bare", "-b", "main", bare).CombinedOutput(); err != nil { - t.Fatalf("git init bare: %v\n%s", err, out) - } - - sandbox := t.TempDir() - initGitRepo(t, sandbox) - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "remote", "add", "origin", bare).CombinedOutput(); err != nil { - t.Fatalf("git remote add: %v\n%s", err, out) - } - if out, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "push", "origin", "main").CombinedOutput(); err != nil { - t.Fatalf("git push initial: %v\n%s", err, out) - } - - headOut, err := exec.Command("git", "-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "rev-parse", "HEAD").Output() - if err != nil { - t.Fatalf("rev-parse HEAD: %v", err) - } - startHEAD := strings.TrimSpace(string(headOut)) - - // Simulate Claude explicitly committing changes. - if err := os.WriteFile(filepath.Join(sandbox, "work.txt"), []byte("done"), 0644); err != nil { - t.Fatal(err) - } - for _, args := range [][]string{ - {"-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "add", "-A"}, - {"-c", "safe.directory=*", "-c", "commit.gpgsign=false", "-C", sandbox, "commit", "-m", "feat: implement the feature"}, - } { - if out, err := exec.Command("git", args...).CombinedOutput(); err != nil { - t.Fatalf("git %v: %v\n%s", args, err, out) - } - } - - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - execRecord := &storage.Execution{} - - if err := teardownSandbox("", sandbox, startHEAD, logger, execRecord); err != nil { - t.Fatalf("teardownSandbox: %v", err) - } - - if len(execRecord.Commits) == 0 { - t.Fatal("expected commits to be captured in execRecord") - } - if !strings.Contains(execRecord.Commits[0].Message, "feat: implement the feature") { - t.Errorf("unexpected commit message: %q", execRecord.Commits[0].Message) - } - if execRecord.Commits[0].Hash == "" { - t.Error("commit hash should not be empty") - } -} - -func TestTeardownSandbox_CleanSandboxWithNoNewCommits_RemovesSandbox(t *testing.T) { - src := t.TempDir() - initGitRepo(t, src) - sandbox, err := setupSandbox(src, slog.Default()) - if err != nil { - t.Fatalf("setupSandbox: %v", err) - } - - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - execRecord := &storage.Execution{} - - headOut, _ := exec.Command("git", "-C", sandbox, "rev-parse", "HEAD").Output() - startHEAD := strings.TrimSpace(string(headOut)) - - // Sandbox has no new commits beyond origin; teardown should succeed and remove it. - if err := teardownSandbox(src, sandbox, startHEAD, logger, execRecord); err != nil { - t.Fatalf("teardownSandbox: %v", err) - } - if _, statErr := os.Stat(sandbox); !os.IsNotExist(statErr) { - t.Error("sandbox should have been removed after clean teardown") - os.RemoveAll(sandbox) - } -} - - -// TestClaudeRunner_Run_ResumeUsesStoredSandboxDir verifies that when a resume -// execution has SandboxDir set, the runner uses that directory (not project_dir) -// as the working directory, so Claude finds its session files there. -func TestClaudeRunner_Run_ResumeUsesStoredSandboxDir(t *testing.T) { - logDir := t.TempDir() - sandboxDir := t.TempDir() - cwdFile := filepath.Join(logDir, "cwd.txt") - - // Use a script that writes its working directory to a file in logDir (stable path). - scriptPath := filepath.Join(t.TempDir(), "fake-claude.sh") - script := "#!/bin/sh\nprintf '%s' \"$PWD\" > " + cwdFile + "\n" - if err := os.WriteFile(scriptPath, []byte(script), 0755); err != nil { - t.Fatalf("write script: %v", err) - } - - r := &ClaudeRunner{ - BinaryPath: scriptPath, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - ProjectDir: sandboxDir, // must exist; resume overrides it with SandboxDir anyway - SkipPlanning: true, - }, - } - exec := &storage.Execution{ - ID: "resume-exec-uuid", - TaskID: "task-1", - ResumeSessionID: "original-session", - ResumeAnswer: "yes", - SandboxDir: sandboxDir, - } - - _ = r.Run(context.Background(), tk, exec) - - got, err := os.ReadFile(cwdFile) - if err != nil { - t.Fatalf("cwd file not written: %v", err) - } - // The runner should have executed claude in sandboxDir, not in project_dir. - if string(got) != sandboxDir { - t.Errorf("resume working dir: want %q, got %q", sandboxDir, string(got)) - } -} - -func TestClaudeRunner_Run_StaleSandboxDir_ClonesAfresh(t *testing.T) { - logDir := t.TempDir() - projectDir := t.TempDir() - initGitRepo(t, projectDir) - - cwdFile := filepath.Join(logDir, "cwd.txt") - scriptPath := filepath.Join(t.TempDir(), "fake-claude.sh") - script := "#!/bin/sh\nprintf '%s' \"$PWD\" > " + cwdFile + "\n" - if err := os.WriteFile(scriptPath, []byte(script), 0755); err != nil { - t.Fatalf("write script: %v", err) - } - - r := &ClaudeRunner{ - BinaryPath: scriptPath, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "claude", - ProjectDir: projectDir, - SkipPlanning: true, - }, - } - // Point to a sandbox that no longer exists (e.g. /tmp was purged). - staleSandbox := filepath.Join(t.TempDir(), "gone") - e := &storage.Execution{ - ID: "resume-exec-2", - TaskID: "task-2", - ResumeSessionID: "session-abc", - ResumeAnswer: "ok", - SandboxDir: staleSandbox, - } - - if err := r.Run(context.Background(), tk, e); err != nil { - t.Fatalf("Run with stale sandbox: %v", err) - } - - got, err := os.ReadFile(cwdFile) - if err != nil { - t.Fatalf("cwd file not written: %v", err) - } - // Should have run in a fresh sandbox (not the stale path, not the raw projectDir). - // The sandbox is removed after teardown, so we only check what it wasn't. - cwd := string(got) - if cwd == staleSandbox { - t.Error("ran in stale sandbox dir that doesn't exist") - } - if cwd == projectDir { - t.Error("ran directly in project_dir; expected a fresh sandbox clone") - } - // cwd should look like a claudomator sandbox path. - if !strings.Contains(cwd, "claudomator-sandbox-") { - t.Errorf("expected sandbox path, got %q", cwd) - } -} - -func TestTailFile_MissingFile_ReturnsEmpty(t *testing.T) { - got := tailFile("/nonexistent/path/file.log", 10) - if got != "" { - t.Errorf("want empty string for missing file, got %q", got) - } -} - diff --git a/internal/executor/container.go b/internal/executor/container.go index 23e35b3..39f737a 100644 --- a/internal/executor/container.go +++ b/internal/executor/container.go @@ -2,6 +2,7 @@ package executor import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -29,6 +30,9 @@ type ContainerRunner struct { ClaudeConfigDir string // host path to ~/.claude; mounted into container for auth credentials CredentialSyncCmd string // optional path to sync-credentials script for auth-error auto-recovery Store Store // optional; used to look up stories and projects for story-aware cloning + // Registry mints the per-task MCP token; when set, the agent gets an + // mcp-config pointing at the host agent MCP server. + Registry *Registry // Command allows mocking exec.CommandContext for tests. Command func(ctx context.Context, name string, arg ...string) *exec.Cmd } @@ -100,7 +104,7 @@ func (r *ContainerRunner) ensureStoryBranch(ctx context.Context, remoteURL, bran return nil } -func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { var err error repoURL := t.RepositoryURL if repoURL == "" { @@ -237,7 +241,7 @@ func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Exec } // Run container (with auth retry on failure). - runErr := r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch) + runErr := r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch, ch) if runErr != nil && isAuthError(runErr) && r.CredentialSyncCmd != "" { r.Logger.Warn("auth failure detected, syncing credentials and retrying once", "taskID", t.ID) syncOut, syncErr := r.command(ctx, r.CredentialSyncCmd).CombinedOutput() @@ -251,7 +255,7 @@ func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Exec if srcData, readErr := os.ReadFile(filepath.Join(r.ClaudeConfigDir, ".claude.json")); readErr == nil { _ = os.WriteFile(filepath.Join(agentHome, ".claude.json"), srcData, 0644) } - runErr = r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch) + runErr = r.runContainer(ctx, t, e, workspace, agentHome, isResume, storyBranch, ch) } if runErr == nil { @@ -267,7 +271,7 @@ func (r *ContainerRunner) Run(ctx context.Context, t *task.Task, e *storage.Exec // runContainer runs the docker container for the given task and handles log setup, // environment files, instructions, and post-execution git operations. -func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *storage.Execution, workspace, agentHome string, isResume bool, storyBranch string) error { +func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *storage.Execution, workspace, agentHome string, isResume bool, storyBranch string, ch AgentChannel) error { repoURL := t.RepositoryURL image := t.Agent.ContainerImage @@ -311,14 +315,39 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto return fmt.Errorf("writing env file: %w", err) } - // Inject custom instructions via file to avoid CLI length limits + // Per-task MCP back-channel: mint a scoped token bound to this run's channel + // and point the agent at the host agent MCP server. Claude reads an + // --mcp-config file; the gemini CLI auto-discovers servers from its + // ~/.gemini/settings.json (agentHome is the container's $HOME). + mcpEnabled := false + if r.Registry != nil { + token, mintErr := r.Registry.Mint(ch) + if mintErr != nil { + return fmt.Errorf("minting agent mcp token: %w", mintErr) + } + defer r.Registry.Revoke(token) + mcpURL := strings.TrimRight(strings.ReplaceAll(r.APIURL, "localhost", "host.docker.internal"), "/") + "/mcp" + if t.Agent.Type == "gemini" { + if err := writeGeminiMCPSettings(agentHome, mcpURL, token); err != nil { + return fmt.Errorf("writing gemini mcp settings: %w", err) + } + } else if err := writeMCPConfig(workspace, mcpURL, token); err != nil { + return fmt.Errorf("writing mcp config: %w", err) + } + mcpEnabled = true + } + + // Inject custom instructions via file to avoid CLI length limits. When the + // MCP back-channel is active, prepend the planning preamble that points the + // agent at the ask_user/report_summary/spawn_subtask/record_progress tools. + instructions := buildAgentInstructions(t, mcpEnabled) instructionsFile := filepath.Join(workspace, ".claudomator-instructions.txt") - if err := os.WriteFile(instructionsFile, []byte(t.Agent.Instructions), 0644); err != nil { + if err := os.WriteFile(instructionsFile, []byte(instructions), 0644); err != nil { return fmt.Errorf("writing instructions: %w", err) } args := r.buildDockerArgs(workspace, agentHome, e.TaskID) - innerCmd := r.buildInnerCmd(t, e, isResume) + innerCmd := r.buildInnerCmd(t, e, isResume, mcpEnabled) fullArgs := append(args, image) fullArgs = append(fullArgs, innerCmd...) @@ -375,31 +404,13 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto e.SessionID = sessionID } - // Check whether the agent left a question before exiting. - questionFile := filepath.Join(logDir, "question.json") - if data, readErr := os.ReadFile(questionFile); readErr == nil { - os.Remove(questionFile) // consumed - questionJSON := strings.TrimSpace(string(data)) - if isCompletionReport(questionJSON) { - r.Logger.Info("treating question file as completion report", "taskID", e.TaskID) - e.Summary = extractQuestionText(questionJSON) - } else { - if e.SessionID == "" { - r.Logger.Warn("missing session ID; resume will start fresh", "taskID", e.TaskID) - } - return &BlockedError{ - QuestionJSON: questionJSON, - SessionID: e.SessionID, - SandboxDir: workspace, - } + // MCP transport: if the agent called ask_user, the question is buffered on + // the channel. Block so the task resumes with the user's answer. + if q, blocked := channelPendingQuestion(ch); blocked { + if e.SessionID == "" { + r.Logger.Warn("missing session ID; resume will start fresh", "taskID", e.TaskID) } - } - - // Read agent summary if written. - summaryFile := filepath.Join(logDir, "summary.txt") - if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil { - os.Remove(summaryFile) // consumed - e.Summary = strings.TrimSpace(string(summaryData)) + return &BlockedError{QuestionJSON: q, SessionID: e.SessionID, SandboxDir: workspace} } // 5. Post-execution: push changes if successful @@ -467,6 +478,10 @@ func (r *ContainerRunner) buildDockerArgs(workspace, claudeHome, taskID string) "-w", "/workspace", "--env-file", hostEnvFile, "-e", "HOME=/home/agent", + // The container is an isolated sandbox; IS_SANDBOX=1 lets the claude CLI + // honor --permission-mode bypassPermissions even when the agent runs as + // root (buildDockerArgs maps --user to the host uid, which may be 0). + "-e", "IS_SANDBOX=1", "-e", "CLAUDOMATOR_API_URL=" + apiURL, "-e", "CLAUDOMATOR_TASK_ID=" + taskID, "-e", "CLAUDOMATOR_DROP_DIR=" + r.DropsDir, @@ -477,7 +492,71 @@ func (r *ContainerRunner) buildDockerArgs(workspace, claudeHome, taskID string) return args } -func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isResume bool) []string { +// mcpConfigContainerPath is where the agent mcp-config is mounted in-container +// (the workspace is bind-mounted at /workspace). +const mcpConfigContainerPath = "/workspace/.claudomator-mcp.json" + +// buildAgentInstructions returns the agent's instructions, prepending the +// MCP-tool planning preamble when the back-channel is active and planning is +// not skipped. +func buildAgentInstructions(t *task.Task, mcpEnabled bool) string { + instructions := t.Agent.Instructions + if mcpEnabled && !t.Agent.SkipPlanning { + instructions = withPlanningPreamble(instructions) + } + return instructions +} + +// writeMCPConfig writes a claude CLI mcp-config that registers the per-task +// agent MCP server over HTTP with a bearer token. +func writeMCPConfig(workspace, mcpURL, token string) error { + cfg := map[string]any{ + "mcpServers": map[string]any{ + "claudomator": map[string]any{ + "type": "http", + "url": mcpURL, + "headers": map[string]string{"Authorization": "Bearer " + token}, + }, + }, + } + data, err := json.Marshal(cfg) + if err != nil { + return err + } + return os.WriteFile(filepath.Join(workspace, ".claudomator-mcp.json"), data, 0600) +} + +// writeGeminiMCPSettings registers the per-task agent MCP server in the gemini +// CLI's user settings (agentHome/.gemini/settings.json, which is $HOME/.gemini +// in-container). The gemini CLI discovers MCP servers from this file rather than +// a command-line flag; httpUrl selects the streamable-HTTP transport and headers +// carry the bearer token. +// +// NOTE: the on-disk schema follows the gemini-cli mcpServers format, but whether +// the CLI actually invokes these tools in non-interactive (-p) mode has not been +// verified against a live gemini binary — confirm with a spike before relying on +// gemini tool-use in production. +func writeGeminiMCPSettings(agentHome, mcpURL, token string) error { + cfg := map[string]any{ + "mcpServers": map[string]any{ + "claudomator": map[string]any{ + "httpUrl": mcpURL, + "headers": map[string]string{"Authorization": "Bearer " + token}, + }, + }, + } + data, err := json.Marshal(cfg) + if err != nil { + return err + } + dir := filepath.Join(agentHome, ".gemini") + if err := os.MkdirAll(dir, 0755); err != nil { + return err + } + return os.WriteFile(filepath.Join(dir, "settings.json"), data, 0600) +} + +func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isResume, mcpEnabled bool) []string { // Claude CLI uses -p for prompt text. To pass a file, we use a shell to cat it. // We use a shell variable to capture the expansion to avoid quoting issues with instructions contents. // The outer single quotes around the sh -c argument prevent host-side expansion. @@ -501,6 +580,9 @@ func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isRe if isResume && e.ResumeSessionID != "" { claudeCmd.WriteString(fmt.Sprintf(" --resume %s", e.ResumeSessionID)) } + if mcpEnabled { + claudeCmd.WriteString(" --mcp-config " + mcpConfigContainerPath) + } claudeCmd.WriteString(" --output-format stream-json --verbose --permission-mode bypassPermissions") return []string{"sh", "-c", claudeCmd.String()} @@ -511,6 +593,7 @@ func (r *ContainerRunner) buildInnerCmd(t *task.Task, e *storage.Execution, isRe var scaffoldPrefixes = []string{ ".claudomator-env", ".claudomator-instructions.txt", + ".claudomator-mcp.json", ".agent-home", } diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go index 54b4689..dec666e 100644 --- a/internal/executor/container_test.go +++ b/internal/executor/container_test.go @@ -2,6 +2,7 @@ package executor import ( "context" + "encoding/json" "fmt" "io" "log/slog" @@ -36,6 +37,7 @@ func TestContainerRunner_BuildDockerArgs(t *testing.T) { "-w", "/workspace", "--env-file", "/tmp/ws/.claudomator-env", "-e", "HOME=/home/agent", + "-e", "IS_SANDBOX=1", "-e", "CLAUDOMATOR_API_URL=http://host.docker.internal:8484", "-e", "CLAUDOMATOR_TASK_ID=task-123", "-e", "CLAUDOMATOR_DROP_DIR=/data/drops", @@ -53,13 +55,96 @@ func TestContainerRunner_BuildDockerArgs(t *testing.T) { } } +func TestBuildAgentInstructions(t *testing.T) { + t.Run("mcp enabled prepends preamble", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Instructions: "do the thing"}} + got := buildAgentInstructions(tk, true) + if !strings.HasPrefix(got, planningPreamble) || !strings.HasSuffix(got, "do the thing") { + t.Errorf("expected preamble + instructions, got %q", got) + } + }) + t.Run("mcp disabled keeps raw instructions", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Instructions: "do the thing"}} + if got := buildAgentInstructions(tk, false); got != "do the thing" { + t.Errorf("expected raw instructions, got %q", got) + } + }) + t.Run("skip planning keeps raw instructions even with mcp", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Instructions: "do the thing", SkipPlanning: true}} + if got := buildAgentInstructions(tk, true); got != "do the thing" { + t.Errorf("expected raw instructions when skip_planning, got %q", got) + } + }) +} + +func TestWriteMCPConfig(t *testing.T) { + dir := t.TempDir() + if err := writeMCPConfig(dir, "http://host.docker.internal:8484/mcp", "tok-abc"); err != nil { + t.Fatalf("writeMCPConfig: %v", err) + } + data, err := os.ReadFile(filepath.Join(dir, ".claudomator-mcp.json")) + if err != nil { + t.Fatalf("read config: %v", err) + } + var parsed struct { + MCPServers map[string]struct { + Type string `json:"type"` + URL string `json:"url"` + Headers map[string]string `json:"headers"` + } `json:"mcpServers"` + } + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("config is not valid JSON: %v", err) + } + srv, ok := parsed.MCPServers["claudomator"] + if !ok { + t.Fatal("expected claudomator server entry") + } + if srv.Type != "http" || srv.URL != "http://host.docker.internal:8484/mcp" { + t.Errorf("unexpected server config: %+v", srv) + } + if srv.Headers["Authorization"] != "Bearer tok-abc" { + t.Errorf("expected bearer header, got %q", srv.Headers["Authorization"]) + } +} + +func TestWriteGeminiMCPSettings(t *testing.T) { + agentHome := t.TempDir() + if err := writeGeminiMCPSettings(agentHome, "http://host.docker.internal:8484/mcp", "tok-xyz"); err != nil { + t.Fatalf("writeGeminiMCPSettings: %v", err) + } + data, err := os.ReadFile(filepath.Join(agentHome, ".gemini", "settings.json")) + if err != nil { + t.Fatalf("read settings: %v", err) + } + var parsed struct { + MCPServers map[string]struct { + HTTPURL string `json:"httpUrl"` + Headers map[string]string `json:"headers"` + } `json:"mcpServers"` + } + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("settings is not valid JSON: %v", err) + } + srv, ok := parsed.MCPServers["claudomator"] + if !ok { + t.Fatal("expected claudomator server entry") + } + if srv.HTTPURL != "http://host.docker.internal:8484/mcp" { + t.Errorf("unexpected httpUrl: %q", srv.HTTPURL) + } + if srv.Headers["Authorization"] != "Bearer tok-xyz" { + t.Errorf("expected bearer header, got %q", srv.Headers["Authorization"]) + } +} + func TestContainerRunner_BuildInnerCmd(t *testing.T) { runner := &ContainerRunner{} t.Run("claude-fresh", func(t *testing.T) { tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} exec := &storage.Execution{} - cmd := runner.buildInnerCmd(tk, exec, false) + cmd := runner.buildInnerCmd(tk, exec, false, false) cmdStr := strings.Join(cmd, " ") if strings.Contains(cmdStr, "--resume") { @@ -73,7 +158,7 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) { t.Run("claude-resume", func(t *testing.T) { tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} exec := &storage.Execution{ResumeSessionID: "orig-session-123"} - cmd := runner.buildInnerCmd(tk, exec, true) + cmd := runner.buildInnerCmd(tk, exec, true, false) cmdStr := strings.Join(cmd, " ") if !strings.Contains(cmdStr, "--resume orig-session-123") { @@ -81,10 +166,26 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) { } }) + t.Run("claude-mcp-enabled", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} + cmdStr := strings.Join(runner.buildInnerCmd(tk, &storage.Execution{}, false, true), " ") + if !strings.Contains(cmdStr, "--mcp-config "+mcpConfigContainerPath) { + t.Errorf("expected --mcp-config flag when MCP enabled, got %q", cmdStr) + } + }) + + t.Run("claude-mcp-disabled", func(t *testing.T) { + tk := &task.Task{Agent: task.AgentConfig{Type: "claude"}} + cmdStr := strings.Join(runner.buildInnerCmd(tk, &storage.Execution{}, false, false), " ") + if strings.Contains(cmdStr, "--mcp-config") { + t.Errorf("did not expect --mcp-config flag when MCP disabled, got %q", cmdStr) + } + }) + t.Run("gemini", func(t *testing.T) { tk := &task.Task{Agent: task.AgentConfig{Type: "gemini"}} exec := &storage.Execution{} - cmd := runner.buildInnerCmd(tk, exec, false) + cmd := runner.buildInnerCmd(tk, exec, false, false) cmdStr := strings.Join(cmd, " ") if !strings.Contains(cmdStr, "gemini -p \"$INST\"") { @@ -99,13 +200,13 @@ func TestContainerRunner_BuildInnerCmd(t *testing.T) { } tkClaude := &task.Task{Agent: task.AgentConfig{Type: "claude"}} - cmdClaude := runnerCustom.buildInnerCmd(tkClaude, &storage.Execution{}, false) + cmdClaude := runnerCustom.buildInnerCmd(tkClaude, &storage.Execution{}, false, false) if !strings.Contains(strings.Join(cmdClaude, " "), "/usr/bin/claude-v2 -p") { t.Errorf("expected custom claude binary, got %q", cmdClaude) } tkGemini := &task.Task{Agent: task.AgentConfig{Type: "gemini"}} - cmdGemini := runnerCustom.buildInnerCmd(tkGemini, &storage.Execution{}, false) + cmdGemini := runnerCustom.buildInnerCmd(tkGemini, &storage.Execution{}, false, false) if !strings.Contains(strings.Join(cmdGemini, " "), "/usr/local/bin/gemini-pro -p") { t.Errorf("expected custom gemini binary, got %q", cmdGemini) } @@ -139,7 +240,7 @@ func TestContainerRunner_Run_PreservesWorkspaceOnFailure(t *testing.T) { } exec := &storage.Execution{ID: "test-exec", TaskID: "test-task"} - err := runner.Run(context.Background(), tk, exec) + err := runner.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID)) if err == nil { t.Fatal("expected error due to mocked docker failure") } @@ -168,48 +269,6 @@ func TestBlockedError_IncludesSandboxDir(t *testing.T) { } } -func TestIsCompletionReport(t *testing.T) { - tests := []struct { - name string - json string - expected bool - }{ - { - name: "real question with options", - json: `{"text": "Should I proceed with implementation?", "options": ["Yes", "No"]}`, - expected: false, - }, - { - name: "real question no options", - json: `{"text": "Which approach do you prefer?"}`, - expected: false, - }, - { - name: "completion report no options no question mark", - json: `{"text": "All tests pass. Implementation complete. Summary written to CLAUDOMATOR_SUMMARY_FILE."}`, - expected: true, - }, - { - name: "completion report with empty options", - json: `{"text": "Feature implemented and committed.", "options": []}`, - expected: true, - }, - { - name: "invalid json treated as not a report", - json: `not json`, - expected: false, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := isCompletionReport(tt.json) - if got != tt.expected { - t.Errorf("isCompletionReport(%q) = %v, want %v", tt.json, got, tt.expected) - } - }) - } -} - func TestTailFile_ReturnsLastNLines(t *testing.T) { f, err := os.CreateTemp("", "tailfile-*") if err != nil { @@ -378,7 +437,7 @@ func TestContainerRunner_MissingCredentials_FailsFast(t *testing.T) { } e := &storage.Execution{ID: "test-exec", TaskID: "test-missing-creds"} - err := runner.Run(context.Background(), tk, e) + err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) if err == nil { t.Fatal("expected error due to missing credentials, got nil") } @@ -418,7 +477,7 @@ func TestContainerRunner_MissingSettings_FailsFast(t *testing.T) { } e := &storage.Execution{ID: "test-exec-2", TaskID: "test-missing-settings"} - err := runner.Run(context.Background(), tk, e) + err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) if err == nil { t.Fatal("expected error due to missing settings, got nil") } @@ -504,7 +563,7 @@ func TestContainerRunner_AuthError_SyncsAndRetries(t *testing.T) { e := &storage.Execution{ID: "auth-retry-exec", TaskID: "auth-retry-test"} // Run — first attempt will fail with auth error, triggering sync+retry - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) // We don't check error strictly since second run may also fail (git push etc.) // What we care about is that docker was called twice and sync was called if callCount < 2 { @@ -550,7 +609,7 @@ func TestContainerRunner_ClonesStoryBranch(t *testing.T) { } e := &storage.Execution{ID: "exec-1", TaskID: "story-branch-test"} - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) os.RemoveAll(e.SandboxDir) // Assert git checkout was called with the story branch name. @@ -597,7 +656,7 @@ func TestContainerRunner_ClonesDefaultBranchWhenNoBranchName(t *testing.T) { } e := &storage.Execution{ID: "exec-2", TaskID: "no-branch-test"} - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)) os.RemoveAll(e.SandboxDir) for _, a := range cloneArgs { @@ -663,7 +722,7 @@ func TestContainerRunner_ClonesFromLocalPathForCITask(t *testing.T) { Agent: task.AgentConfig{Type: "claude"}, } e := &storage.Execution{ID: "exec-ci-1", TaskID: "ci-task-1"} - runner.Run(context.Background(), tk, e) + runner.Run(context.Background(), tk, e, noopChannel{}) os.RemoveAll(e.SandboxDir) if cloneSrc == "https://github.com/thepeterstone/nav.git" { @@ -752,3 +811,13 @@ func TestEnsureStoryBranch_IdempotentIfExists(t *testing.T) { t.Fatalf("ensureStoryBranch on existing branch: %v", err) } } + +// noopChannel satisfies AgentChannel with no-op implementations for tests. +type noopChannel struct{} + +func (noopChannel) AskUser(_ context.Context, _ string) (string, error) { return "", nil } +func (noopChannel) ReportSummary(_ context.Context, _ string) error { return nil } +func (noopChannel) SpawnSubtask(_ context.Context, _ SubtaskSpec) (string, error) { + return "", nil +} +func (noopChannel) RecordProgress(_ context.Context, _ string) error { return nil } diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 09169bd..8614481 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -12,6 +12,7 @@ import ( "sync" "time" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/retry" "github.com/thepeterstone/claudomator/internal/storage" @@ -41,6 +42,7 @@ type Store interface { ListTasksByStory(storyID string) ([]*task.Task, error) UpdateStoryStatus(id string, status task.StoryState) error CreateTask(t *task.Task) error + CreateEvent(e *event.Event) error UpdateTaskCheckerReport(id, report string) error GetCheckerTask(checkedTaskID string) (*task.Task, error) } @@ -52,9 +54,11 @@ type LogPather interface { ExecLogDir(execID string) string } -// Runner executes a single task and returns the result. +// Runner executes a single task and returns the result. The AgentChannel is how +// the runner reports agent-originated signals (summary, questions, subtasks, +// progress) back to the system. type Runner interface { - Run(ctx context.Context, t *task.Task, exec *storage.Execution) error + Run(ctx context.Context, t *task.Task, exec *storage.Execution, ch AgentChannel) error } // workItem is an entry in the pool's internal work queue. @@ -89,6 +93,14 @@ type Pool struct { dispatchDone chan struct{} // closed when the dispatch goroutine exits Classifier *Classifier LLM *llm.Client + // Budget gates paid providers against a rolling window; nil disables gating. + Budget BudgetGate +} + +// BudgetGate reports whether a task estimated at estCost may run on a provider +// without breaching its rolling spend window. Satisfied by *budget.Accountant. +type BudgetGate interface { + Allow(provider string, estCost float64) (bool, error) } // Result is emitted when a task execution completes. @@ -352,8 +364,12 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex } } - err = runner.Run(ctx, t, exec) + sc := newStoreChannel(p.store, t.ID) + err = runner.Run(ctx, t, exec, sc) exec.EndTime = time.Now().UTC() + if sum, ok := sc.ReportedSummary(); ok { + exec.Summary = sum + } p.decActiveAgent(agentType, &cleaned) p.handleRunResult(ctx, t, exec, err, agentType) @@ -929,6 +945,44 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { agentType = "claude" } + // Budget gating: if running on this provider could breach its rolling window + // (estimated by the task's max_budget_usd), reroute to the free local runner + // when one is registered, otherwise stop and mark the task BUDGET_EXCEEDED. + if p.Budget != nil { + if ok, berr := p.Budget.Allow(agentType, t.Agent.MaxBudgetUSD); berr != nil { + p.logger.Warn("budget check failed; proceeding", "error", berr, "taskID", t.ID) + } else if !ok { + if _, hasLocal := p.runners["local"]; hasLocal && agentType != "local" { + p.logger.Info("budget window would be breached; rerouting to local", "taskID", t.ID, "from", agentType) + t.Agent.Type = "local" + t.Agent.Model = "" + agentType = "local" + if uerr := p.store.UpdateTaskAgent(t.ID, t.Agent); uerr != nil { + p.logger.Error("failed to persist rerouted agent", "error", uerr, "taskID", t.ID) + } + } else { + now := time.Now().UTC() + exec := &storage.Execution{ + ID: uuid.New().String(), + TaskID: t.ID, + StartTime: now, + EndTime: now, + Status: "BUDGET_EXCEEDED", + Agent: agentType, + ErrorMsg: fmt.Sprintf("rolling budget window cap reached for provider %q", agentType), + } + if createErr := p.store.CreateExecution(exec); createErr != nil { + p.logger.Error("failed to create execution record", "error", createErr) + } + if err := p.store.UpdateTaskState(t.ID, task.StateBudgetExceeded); err != nil { + p.logger.Error("failed to update task state", "taskID", t.ID, "state", task.StateBudgetExceeded, "error", err) + } + p.resultCh <- &Result{TaskID: t.ID, Execution: exec, Err: fmt.Errorf("budget cap reached for %s", agentType)} + return + } + } + } + // Check dependencies before taking the per-agent slot to avoid deadlock: // if a dependent task holds the slot while waiting for its dependency to run, // the dependency can never start (maxPerAgent=1). @@ -1017,6 +1071,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { TaskID: t.ID, StartTime: time.Now().UTC(), Status: "RUNNING", + Agent: agentType, } // Pre-populate log paths so they're available in the DB immediately — @@ -1073,8 +1128,12 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { } // Run the task. - err = runner.Run(ctx, t, exec) + sc := newStoreChannel(p.store, t.ID) + err = runner.Run(ctx, t, exec, sc) exec.EndTime = time.Now().UTC() + if sum, ok := sc.ReportedSummary(); ok { + exec.Summary = sum + } p.decActiveAgent(agentType, &cleaned) p.handleRunResult(ctx, t, exec, err, agentType) diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index 9214872..b737e22 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" ) @@ -63,7 +64,7 @@ type mockRunner struct { onRun func(*task.Task, *storage.Execution) error } -func (m *mockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (m *mockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, _ AgentChannel) error { m.mu.Lock() m.calls++ cb := m.onRun @@ -216,6 +217,63 @@ func TestPool_Submit_Subtask_GoesToCompleted(t *testing.T) { } } +type fakeGate struct{ deny map[string]bool } + +func (g fakeGate) Allow(provider string, _ float64) (bool, error) { return !g.deny[provider], nil } + +func TestPool_BudgetGate_BlocksWhenNoLocalFallback(t *testing.T) { + store := testStore(t) + runner := &mockRunner{} + pool := NewPool(2, map[string]Runner{"claude": runner}, store, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))) + pool.Budget = fakeGate{deny: map[string]bool{"claude": true}} + + tk := makeTask("bg-1") + store.CreateTask(tk) + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("submit: %v", err) + } + + result := <-pool.Results() + if result.Execution.Status != "BUDGET_EXCEEDED" { + t.Errorf("status: want BUDGET_EXCEEDED, got %q", result.Execution.Status) + } + if runner.callCount() != 0 { + t.Errorf("claude runner should not have run, got %d calls", runner.callCount()) + } + got, _ := store.GetTask("bg-1") + if got.State != task.StateBudgetExceeded { + t.Errorf("task state: want BUDGET_EXCEEDED, got %v", got.State) + } +} + +func TestPool_BudgetGate_ReroutesToLocal(t *testing.T) { + store := testStore(t) + claudeRunner := &mockRunner{} + localRunner := &mockRunner{} + pool := NewPool(2, map[string]Runner{"claude": claudeRunner, "local": localRunner}, store, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))) + pool.Budget = fakeGate{deny: map[string]bool{"claude": true}} // local is allowed + + tk := makeTask("bg-2") + store.CreateTask(tk) + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("submit: %v", err) + } + + result := <-pool.Results() + if result.Err != nil { + t.Errorf("expected success after reroute, got %v", result.Err) + } + if localRunner.callCount() != 1 { + t.Errorf("local runner should have run once, got %d", localRunner.callCount()) + } + if claudeRunner.callCount() != 0 { + t.Errorf("claude runner should not have run, got %d", claudeRunner.callCount()) + } + if result.Execution.Agent != "local" { + t.Errorf("execution agent: want local, got %q", result.Execution.Agent) + } +} + func TestPool_Submit_Failure(t *testing.T) { store := testStore(t) runner := &mockRunner{err: fmt.Errorf("boom"), exitCode: 1} @@ -391,11 +449,11 @@ func (m *logPatherMockRunner) ExecLogDir(execID string) string { return filepath.Join(m.logDir, execID) } -func (m *logPatherMockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +func (m *logPatherMockRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error { m.mu.Lock() m.capturedPath = e.StdoutPath m.mu.Unlock() - return m.mockRunner.Run(ctx, t, e) + return m.mockRunner.Run(ctx, t, e, ch) } // TestPool_Execute_LogPathsPreSetBeforeRun verifies that when the runner @@ -1194,6 +1252,7 @@ func (m *minimalMockStore) GetStory(_ string) (*task.Story, error) func (m *minimalMockStore) ListTasksByStory(_ string) ([]*task.Task, error) { return nil, nil } func (m *minimalMockStore) UpdateStoryStatus(_ string, _ task.StoryState) error { return nil } func (m *minimalMockStore) CreateTask(_ *task.Task) error { return nil } +func (m *minimalMockStore) CreateEvent(_ *event.Event) error { return nil } func (m *minimalMockStore) UpdateTaskCheckerReport(_ string, _ string) error { return nil } func (m *minimalMockStore) GetCheckerTask(_ string) (*task.Task, error) { return nil, nil } diff --git a/internal/executor/gemini.go b/internal/executor/gemini.go deleted file mode 100644 index 3abec05..0000000 --- a/internal/executor/gemini.go +++ /dev/null @@ -1,346 +0,0 @@ -package executor - -import ( - "context" - "encoding/json" - "fmt" - "io" - "log/slog" - "os" - "os/exec" - "path/filepath" - "strings" - "sync" - "syscall" - - "github.com/thepeterstone/claudomator/internal/storage" - "github.com/thepeterstone/claudomator/internal/task" -) - -// GeminiRunner spawns the `gemini` CLI in non-interactive mode. -type GeminiRunner struct { - BinaryPath string // defaults to "gemini" - Logger *slog.Logger - LogDir string // base directory for execution logs - APIURL string // base URL of the Claudomator API, passed to subprocesses -} - -// ExecLogDir returns the log directory for the given execution ID. -func (r *GeminiRunner) ExecLogDir(execID string) string { - if r.LogDir == "" { - return "" - } - return filepath.Join(r.LogDir, execID) -} - -func (r *GeminiRunner) binaryPath() string { - if r.BinaryPath != "" { - return r.BinaryPath - } - return "gemini" -} - -// Run executes the gemini CLI inside a sandboxed clone of project_dir. -// When project_dir is set, claudomator first clones it into a temp sandbox -// (preferring a `local` bare remote, then `origin`, then the working tree) -// and runs the agent there. On success the sandbox is autocommitted and -// pushed back to origin/master, then removed. On failure the sandbox is -// preserved and its path is included in the returned error so the user can -// inspect partial work. If the agent writes a question file before exiting, -// Run returns *BlockedError with SandboxDir populated so a resume execution -// can pick up in the same directory. -func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { - projectDir := t.Agent.ProjectDir - - if projectDir != "" { - if _, err := os.Stat(projectDir); err != nil { - return fmt.Errorf("project_dir %q: %w", projectDir, err) - } - } - - logDir := r.ExecLogDir(e.ID) - if logDir == "" { - logDir = e.ID - } - if err := os.MkdirAll(logDir, 0700); err != nil { - return fmt.Errorf("creating log dir: %w", err) - } - - if e.StdoutPath == "" { - e.StdoutPath = filepath.Join(logDir, "stdout.log") - e.StderrPath = filepath.Join(logDir, "stderr.log") - e.ArtifactDir = logDir - } - - if e.SessionID == "" { - if e.ResumeSessionID != "" { - e.SessionID = e.ResumeSessionID - } else { - e.SessionID = e.ID - } - } - - // Sandbox setup: for new executions with a project_dir, clone into a sandbox. - // Resume executions reuse the preserved sandbox so any partial work survives. - // If the preserved sandbox is missing (e.g. /tmp was purged), clone fresh. - var sandboxDir string - var startHEAD string - effectiveWorkingDir := projectDir - if e.ResumeSessionID != "" { - if e.SandboxDir != "" { - if _, statErr := os.Stat(e.SandboxDir); statErr == nil { - effectiveWorkingDir = e.SandboxDir - } else { - r.Logger.Warn("preserved sandbox missing, cloning fresh", "sandbox", e.SandboxDir, "project_dir", projectDir) - e.SandboxDir = "" - if projectDir != "" { - var err error - sandboxDir, err = setupSandbox(projectDir, r.Logger) - if err != nil { - return fmt.Errorf("setting up sandbox: %w", err) - } - effectiveWorkingDir = sandboxDir - r.Logger.Info("fresh sandbox created for resume", "sandbox", sandboxDir, "project_dir", projectDir) - } - } - } - } else if projectDir != "" { - var err error - sandboxDir, err = setupSandbox(projectDir, r.Logger) - if err != nil { - return fmt.Errorf("setting up sandbox: %w", err) - } - effectiveWorkingDir = sandboxDir - r.Logger.Info("sandbox created", "sandbox", sandboxDir, "project_dir", projectDir) - } - - if effectiveWorkingDir != "" { - headOut, _ := exec.Command("git", gitSafe("-C", effectiveWorkingDir, "rev-parse", "HEAD")...).Output() - startHEAD = strings.TrimSpace(string(headOut)) - } - - questionFile := filepath.Join(logDir, "question.json") - args := r.buildArgs(t, e, questionFile) - - if err := r.execOnce(ctx, args, effectiveWorkingDir, projectDir, e); err != nil { - if sandboxDir != "" { - return fmt.Errorf("%w (sandbox preserved at %s)", err, sandboxDir) - } - return err - } - - // Check whether the agent left a question before exiting. - data, readErr := os.ReadFile(questionFile) - if readErr == nil { - os.Remove(questionFile) - questionJSON := strings.TrimSpace(string(data)) - if isCompletionReport(questionJSON) { - r.Logger.Info("treating question file as completion report", "taskID", e.TaskID) - e.Summary = extractQuestionText(questionJSON) - } else { - // Preserve sandbox on BLOCKED so a resume can pick up in the same dir. - return &BlockedError{QuestionJSON: questionJSON, SessionID: e.SessionID, SandboxDir: sandboxDir} - } - } - - // Read agent summary if written. - summaryFile := filepath.Join(logDir, "summary.txt") - if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil { - os.Remove(summaryFile) - e.Summary = strings.TrimSpace(string(summaryData)) - } - - // Merge sandbox back to project_dir and clean up. - if sandboxDir != "" { - if mergeErr := teardownSandbox(projectDir, sandboxDir, startHEAD, r.Logger, e); mergeErr != nil { - return fmt.Errorf("sandbox teardown: %w (sandbox preserved at %s)", mergeErr, sandboxDir) - } - } - return nil -} - -func (r *GeminiRunner) execOnce(ctx context.Context, args []string, workingDir, projectDir string, e *storage.Execution) error { - cmd := exec.CommandContext(ctx, r.binaryPath(), args...) - cmd.Env = append(os.Environ(), - "CLAUDOMATOR_API_URL="+r.APIURL, - "CLAUDOMATOR_TASK_ID="+e.TaskID, - "CLAUDOMATOR_PROJECT_DIR="+projectDir, - "CLAUDOMATOR_QUESTION_FILE="+filepath.Join(e.ArtifactDir, "question.json"), - "CLAUDOMATOR_SUMMARY_FILE="+filepath.Join(e.ArtifactDir, "summary.txt"), - ) - cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} - if workingDir != "" { - cmd.Dir = workingDir - } - - stdoutFile, err := os.Create(e.StdoutPath) - if err != nil { - return fmt.Errorf("creating stdout log: %w", err) - } - defer stdoutFile.Close() - - stderrFile, err := os.Create(e.StderrPath) - if err != nil { - return fmt.Errorf("creating stderr log: %w", err) - } - defer stderrFile.Close() - - stdoutR, stdoutW, err := os.Pipe() - if err != nil { - return fmt.Errorf("creating stdout pipe: %w", err) - } - cmd.Stdout = stdoutW - cmd.Stderr = stderrFile - - if err := cmd.Start(); err != nil { - stdoutW.Close() - stdoutR.Close() - return fmt.Errorf("starting gemini: %w", err) - } - stdoutW.Close() - - killDone := make(chan struct{}) - go func() { - select { - case <-ctx.Done(): - syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) - case <-killDone: - } - }() - - var streamCost float64 - var streamErr error - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - streamCost, streamErr = parseGeminiStream(stdoutR, stdoutFile, r.Logger) - stdoutR.Close() - }() - - waitErr := cmd.Wait() - close(killDone) - wg.Wait() - - if streamCost > 0 { - e.CostUSD = streamCost - } - - if waitErr != nil { - if exitErr, ok := waitErr.(*exec.ExitError); ok { - e.ExitCode = exitErr.ExitCode() - } - if streamErr != nil { - return streamErr - } - if tail := tailFile(e.StderrPath, 20); tail != "" { - return fmt.Errorf("gemini exited with error: %w\nstderr:\n%s", waitErr, tail) - } - return fmt.Errorf("gemini exited with error: %w", waitErr) - } - - if streamErr != nil { - return streamErr - } - return nil -} - -// parseGeminiStream reads streaming JSON from the gemini CLI, strips markdown -// code fences if the output is wrapped in them, writes the inner stream-json -// to w, and returns (costUSD, error). If a `result` event has `is_error: true`, -// an error wrapping the result message is returned. -func parseGeminiStream(r io.Reader, w io.Writer, logger *slog.Logger) (float64, error) { - fullOutput, err := io.ReadAll(r) - if err != nil { - return 0, fmt.Errorf("reading full gemini output: %w", err) - } - logger.Debug("parseGeminiStream: raw output received", "output", string(fullOutput)) - - inner := stripGeminiFences(string(fullOutput), logger) - if _, writeErr := w.Write([]byte(inner)); writeErr != nil { - return 0, fmt.Errorf("writing gemini output: %w", writeErr) - } - - // Walk lines looking for a result event so we can surface errors and cost. - var ( - cost float64 - errMsg string - isError bool - ) - for _, raw := range strings.Split(inner, "\n") { - line := strings.TrimSpace(raw) - if line == "" { - continue - } - var evt struct { - Type string `json:"type"` - IsError bool `json:"is_error"` - Result string `json:"result"` - Cost float64 `json:"total_cost_usd"` - } - if err := json.Unmarshal([]byte(line), &evt); err != nil { - continue - } - if evt.Type == "result" { - if evt.Cost > 0 { - cost = evt.Cost - } - if evt.IsError { - isError = true - errMsg = evt.Result - } - } - } - if isError { - return cost, fmt.Errorf("gemini reported error: %s", errMsg) - } - return cost, nil -} - -// stripGeminiFences removes a surrounding ```json ... ``` markdown block if -// present, returning the trimmed inner content. If no markdown fence is -// found, the input is returned verbatim (no whitespace trimming) so callers -// that expect byte-exact pass-through behavior get it. -func stripGeminiFences(raw string, logger *slog.Logger) string { - trimmed := strings.TrimSpace(raw) - if start := strings.Index(trimmed, "```json"); start != -1 { - if end := strings.LastIndex(trimmed, "```"); end > start { - return strings.TrimSpace(trimmed[start+len("```json") : end]) - } - logger.Warn("malformed gemini markdown block (missing closing fence); using raw output", "len", len(trimmed)) - return trimmed - } - return raw -} - -func (r *GeminiRunner) buildArgs(t *task.Task, e *storage.Execution, questionFile string) []string { - // Gemini CLI uses a different command structure: gemini "instructions" [flags] - - instructions := t.Agent.Instructions - if !t.Agent.SkipPlanning { - instructions = withPlanningPreamble(instructions) - } - - args := []string{ - "-p", instructions, - "--output-format", "stream-json", - "--yolo", // auto-approve all tools (equivalent to Claude's bypassPermissions) - } - - // Note: Gemini CLI flags might differ from Claude CLI. - // Assuming common flags for now, but these may need adjustment. - if t.Agent.Model != "" { - args = append(args, "--model", t.Agent.Model) - } - - // Gemini CLI doesn't use --session-id for the first run in the same way, - // or it might use it differently. For now we assume compatibility. - if e.SessionID != "" { - // If it's a resume, it might use different flags. - if e.ResumeSessionID != "" { - // This is a placeholder for Gemini's resume logic - } - } - - return args -} diff --git a/internal/executor/gemini_test.go b/internal/executor/gemini_test.go deleted file mode 100644 index cd11ebc..0000000 --- a/internal/executor/gemini_test.go +++ /dev/null @@ -1,447 +0,0 @@ -package executor - -import ( - "bytes" - "context" - "errors" - "io" - "log/slog" - "os" - "path/filepath" - "strings" - "testing" - - "github.com/thepeterstone/claudomator/internal/storage" - "github.com/thepeterstone/claudomator/internal/task" -) - -func TestGeminiRunner_BuildArgs_BasicTask(t *testing.T) { - r := &GeminiRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "fix the bug", - Model: "gemini-2.5-flash-lite", - SkipPlanning: true, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - // Gemini CLI: instructions passed via -p for non-interactive mode - if len(args) < 2 || args[0] != "-p" || args[1] != "fix the bug" { - t.Errorf("expected -p <instructions> as first args, got: %v", args) - } - - argMap := make(map[string]bool) - for _, a := range args { - argMap[a] = true - } - for _, want := range []string{"--output-format", "stream-json", "--model", "gemini-2.5-flash-lite"} { - if !argMap[want] { - t.Errorf("missing arg %q in %v", want, args) - } - } -} - -func TestGeminiRunner_BuildArgs_PreamblePrepended(t *testing.T) { - r := &GeminiRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "fix the bug", - SkipPlanning: false, - }, - } - - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - - if len(args) < 2 || args[0] != "-p" { - t.Fatalf("expected -p <instructions> as first args, got: %v", args) - } - if !strings.HasPrefix(args[1], planningPreamble) { - t.Errorf("instructions should start with planning preamble") - } - if !strings.HasSuffix(args[1], "fix the bug") { - t.Errorf("instructions should end with original instructions") - } -} - -func TestGeminiRunner_BuildArgs_IncludesYolo(t *testing.T) { - r := &GeminiRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "write a doc", - SkipPlanning: true, - }, - } - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - argMap := make(map[string]bool) - for _, a := range args { - argMap[a] = true - } - if !argMap["--yolo"] { - t.Errorf("expected --yolo in gemini args (enables all tools); got: %v", args) - } -} - -func TestGeminiRunner_BuildArgs_IncludesPromptFlag(t *testing.T) { - r := &GeminiRunner{} - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "do the thing", - SkipPlanning: true, - }, - } - args := r.buildArgs(tk, &storage.Execution{ID: "test-exec"}, "/tmp/q.json") - // Instructions must be passed via -p/--prompt for non-interactive headless mode, - // not as a bare positional (which starts interactive mode). - found := false - for i, a := range args { - if (a == "-p" || a == "--prompt") && i+1 < len(args) && args[i+1] == "do the thing" { - found = true - break - } - } - if !found { - t.Errorf("expected instructions passed via -p/--prompt flag; got: %v", args) - } -} - -func TestGeminiRunner_Run_InaccessibleProjectDir_ReturnsError(t *testing.T) { - r := &GeminiRunner{ - BinaryPath: "true", // would succeed if it ran - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: t.TempDir(), - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - ProjectDir: "/nonexistent/path/does/not/exist", - SkipPlanning: true, - }, - } - exec := &storage.Execution{ID: "test-exec"} - - err := r.Run(context.Background(), tk, exec) - - if err == nil { - t.Fatal("expected error for inaccessible project_dir, got nil") - } - if !strings.Contains(err.Error(), "project_dir") { - t.Errorf("expected 'project_dir' in error, got: %v", err) - } -} - -func TestGeminiRunner_BinaryPath_Default(t *testing.T) { - r := &GeminiRunner{} - if r.binaryPath() != "gemini" { - t.Errorf("want 'gemini', got %q", r.binaryPath()) - } -} - -func TestGeminiRunner_BinaryPath_Custom(t *testing.T) { - r := &GeminiRunner{BinaryPath: "/usr/local/bin/gemini"} - if r.binaryPath() != "/usr/local/bin/gemini" { - t.Errorf("want custom path, got %q", r.binaryPath()) - } -} - - -func TestParseGeminiStream_ParsesStructuredOutput(t *testing.T) { - // Simulate a stream-json input with various message types, including a result with error and cost. - input := streamLine(`{"type":"content_block_start","content_block":{"text":"Hello,"}}`) + - streamLine(`{"type":"content_block_delta","content_block":{"text":" World!"}}`) + - streamLine(`{"type":"content_block_end"}`) + - streamLine(`{"type":"result","subtype":"error_during_execution","is_error":true,"result":"something went wrong","total_cost_usd":0.123}`) - - reader := strings.NewReader(input) - var writer bytes.Buffer // To capture what's written to the output log - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - - cost, err := parseGeminiStream(reader, &writer, logger) - - if err == nil { - t.Errorf("expected an error, got nil") - } - if !strings.Contains(err.Error(), "something went wrong") { - t.Errorf("expected error message to contain 'something went wrong', got: %v", err) - } - - if cost != 0.123 { - t.Errorf("expected cost 0.123, got %f", cost) - } - - // Verify that the writer received the content (even if parseGeminiStream isn't fully parsing it yet) - expectedWriterContent := input - if writer.String() != expectedWriterContent { - t.Errorf("writer content mismatch:\nwant:\n%s\ngot:\n%s", expectedWriterContent, writer.String()) - } -} - -// TestGeminiRunner_Run_ProjectDir_RunsInSandbox verifies that when project_dir -// is set, the gemini subprocess runs inside a sandbox clone — not in -// project_dir itself. -func TestGeminiRunner_Run_ProjectDir_RunsInSandbox(t *testing.T) { - projectDir := t.TempDir() - initGitRepo(t, projectDir) - - logDir := t.TempDir() - cwdFile := filepath.Join(logDir, "gemini-cwd.txt") - - // Fake gemini binary that records its $PWD then exits 0. - scriptPath := filepath.Join(t.TempDir(), "fake-gemini.sh") - script := "#!/bin/sh\nprintf '%s' \"$PWD\" > " + cwdFile + "\n" - if err := os.WriteFile(scriptPath, []byte(script), 0755); err != nil { - t.Fatalf("write script: %v", err) - } - - r := &GeminiRunner{ - BinaryPath: scriptPath, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "do work", - ProjectDir: projectDir, - SkipPlanning: true, - }, - } - e := &storage.Execution{ID: "sandbox-exec", TaskID: "task-1"} - - if err := r.Run(context.Background(), tk, e); err != nil { - t.Fatalf("Run: %v", err) - } - - got, err := os.ReadFile(cwdFile) - if err != nil { - t.Fatalf("cwd file not written: %v", err) - } - cwd := string(got) - if cwd == projectDir { - t.Errorf("ran directly in project_dir; expected sandbox clone (cwd=%q)", cwd) - } - // Sandbox should be removed after successful teardown (no edits → nothing to push). - // We can't assert the exact dir, but it should not be projectDir. -} - -// TestGeminiRunner_Run_BlockedError_IncludesSandboxDir verifies that when the -// agent writes a question file before exiting, the BlockedError carries the -// sandbox path so resume runs in the same dir. -func TestGeminiRunner_Run_BlockedError_IncludesSandboxDir(t *testing.T) { - src := t.TempDir() - initGitRepo(t, src) - logDir := t.TempDir() - - scriptPath := filepath.Join(t.TempDir(), "fake-gemini.sh") - if err := os.WriteFile(scriptPath, []byte(`#!/bin/sh -if [ -n "$CLAUDOMATOR_QUESTION_FILE" ]; then - printf '{"text":"Should I continue?"}' > "$CLAUDOMATOR_QUESTION_FILE" -fi -`), 0755); err != nil { - t.Fatalf("write script: %v", err) - } - - r := &GeminiRunner{ - BinaryPath: scriptPath, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "do something", - ProjectDir: src, - SkipPlanning: true, - }, - } - e := &storage.Execution{ID: "blocked-gemini-exec", TaskID: "task-1"} - - err := r.Run(context.Background(), tk, e) - - var blocked *BlockedError - if !errors.As(err, &blocked) { - t.Fatalf("expected BlockedError, got: %v", err) - } - if blocked.SandboxDir == "" { - t.Error("BlockedError.SandboxDir should be set when gemini task runs in a sandbox") - } - if _, statErr := os.Stat(blocked.SandboxDir); os.IsNotExist(statErr) { - t.Error("sandbox directory should be preserved when blocked") - } else { - os.RemoveAll(blocked.SandboxDir) - } -} - -// TestGeminiRunner_Run_ExecError_PreservesSandbox verifies that when gemini -// exits non-zero, the sandbox path is included in the wrapped error so the -// user can inspect partial work. -func TestGeminiRunner_Run_ExecError_PreservesSandbox(t *testing.T) { - src := t.TempDir() - initGitRepo(t, src) - logDir := t.TempDir() - - // "false" exits 1, no output. - r := &GeminiRunner{ - BinaryPath: "false", - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "do something", - ProjectDir: src, - SkipPlanning: true, - }, - } - e := &storage.Execution{ID: "err-gemini-exec", TaskID: "task-1"} - - err := r.Run(context.Background(), tk, e) - if err == nil { - t.Fatal("expected error from failing gemini exit") - } - if !strings.Contains(err.Error(), "sandbox preserved at ") { - t.Errorf("expected error to include sandbox path; got: %v", err) - } - // Extract path and verify it exists. - idx := strings.Index(err.Error(), "sandbox preserved at ") - rest := err.Error()[idx+len("sandbox preserved at "):] - rest = strings.TrimSuffix(rest, ")") - rest = strings.TrimSpace(rest) - if _, statErr := os.Stat(rest); os.IsNotExist(statErr) { - t.Errorf("sandbox path from error should exist on disk: %q", rest) - } else { - os.RemoveAll(rest) - } -} - -// TestGeminiRunner_Run_ResumeUsesStoredSandboxDir verifies that a resume -// execution runs in the preserved SandboxDir rather than cloning fresh. -func TestGeminiRunner_Run_ResumeUsesStoredSandboxDir(t *testing.T) { - logDir := t.TempDir() - sandboxDir := t.TempDir() - initGitRepo(t, sandboxDir) - cwdFile := filepath.Join(logDir, "cwd.txt") - - scriptPath := filepath.Join(t.TempDir(), "fake-gemini.sh") - script := "#!/bin/sh\nprintf '%s' \"$PWD\" > " + cwdFile + "\n" - if err := os.WriteFile(scriptPath, []byte(script), 0755); err != nil { - t.Fatalf("write script: %v", err) - } - - r := &GeminiRunner{ - BinaryPath: scriptPath, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - SkipPlanning: true, - }, - } - e := &storage.Execution{ - ID: "resume-gemini-1", - TaskID: "task-resume", - ResumeSessionID: "session-abc", - SandboxDir: sandboxDir, - } - - if err := r.Run(context.Background(), tk, e); err != nil { - t.Fatalf("Run with preserved sandbox: %v", err) - } - - got, err := os.ReadFile(cwdFile) - if err != nil { - t.Fatalf("cwd file not written: %v", err) - } - if string(got) != sandboxDir { - t.Errorf("resume should run in preserved sandbox; got cwd=%q want %q", got, sandboxDir) - } -} - -// TestGeminiRunner_Run_StaleSandboxDir_ClonesAfresh verifies that a resume -// pointing at a missing sandbox falls back to cloning a fresh sandbox from -// project_dir rather than failing outright. -func TestGeminiRunner_Run_StaleSandboxDir_ClonesAfresh(t *testing.T) { - logDir := t.TempDir() - projectDir := t.TempDir() - initGitRepo(t, projectDir) - - cwdFile := filepath.Join(logDir, "cwd.txt") - scriptPath := filepath.Join(t.TempDir(), "fake-gemini.sh") - script := "#!/bin/sh\nprintf '%s' \"$PWD\" > " + cwdFile + "\n" - if err := os.WriteFile(scriptPath, []byte(script), 0755); err != nil { - t.Fatalf("write script: %v", err) - } - - r := &GeminiRunner{ - BinaryPath: scriptPath, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - ProjectDir: projectDir, - SkipPlanning: true, - }, - } - staleSandbox := filepath.Join(t.TempDir(), "gone") - e := &storage.Execution{ - ID: "resume-gemini-2", - TaskID: "task-stale", - ResumeSessionID: "session-xyz", - SandboxDir: staleSandbox, - } - - if err := r.Run(context.Background(), tk, e); err != nil { - t.Fatalf("Run with stale sandbox: %v", err) - } - - got, err := os.ReadFile(cwdFile) - if err != nil { - t.Fatalf("cwd file not written: %v", err) - } - cwd := string(got) - if cwd == staleSandbox { - t.Error("ran in stale (nonexistent) sandbox dir") - } - if cwd == projectDir { - t.Error("ran directly in project_dir; expected a fresh sandbox clone") - } -} - -// TestGeminiRunner_Run_NoProjectDir_SkipsSandbox verifies that a task with no -// project_dir doesn't trigger sandbox setup (matches LocalRunner/non-coding -// task semantics). -func TestGeminiRunner_Run_NoProjectDir_SkipsSandbox(t *testing.T) { - logDir := t.TempDir() - - r := &GeminiRunner{ - BinaryPath: "true", // exits 0, no output - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logDir, - } - tk := &task.Task{ - Agent: task.AgentConfig{ - Type: "gemini", - Instructions: "summarize: 2+2", - SkipPlanning: true, - // No ProjectDir - }, - } - e := &storage.Execution{ID: "no-pd-gemini", TaskID: "task-nopd"} - - if err := r.Run(context.Background(), tk, e); err != nil { - t.Fatalf("Run without project_dir: %v", err) - } - if e.SandboxDir != "" { - t.Errorf("SandboxDir should be empty for tasks without project_dir, got %q", e.SandboxDir) - } -} diff --git a/internal/executor/helpers.go b/internal/executor/helpers.go index 76bf8b1..9517492 100644 --- a/internal/executor/helpers.go +++ b/internal/executor/helpers.go @@ -177,29 +177,3 @@ func gitSafe(args ...string) []string { "-c", "tag.gpgsign=false", }, args...) } - -// isCompletionReport returns true when a question-file JSON looks like a -// completion report rather than a real user question. Heuristic: no options -// (or empty options) and no "?" anywhere in the text. -func isCompletionReport(questionJSON string) bool { - var q struct { - Text string `json:"text"` - Options []string `json:"options"` - } - if err := json.Unmarshal([]byte(questionJSON), &q); err != nil { - return false - } - return len(q.Options) == 0 && !strings.Contains(q.Text, "?") -} - -// extractQuestionText returns the "text" field from a question-file JSON, or -// the raw string if parsing fails. -func extractQuestionText(questionJSON string) string { - var q struct { - Text string `json:"text"` - } - if err := json.Unmarshal([]byte(questionJSON), &q); err != nil { - return questionJSON - } - return strings.TrimSpace(q.Text) -} diff --git a/internal/executor/local.go b/internal/executor/local.go index 5d874c6..3f20fe7 100644 --- a/internal/executor/local.go +++ b/internal/executor/local.go @@ -37,10 +37,17 @@ func (r *LocalRunner) ExecLogDir(execID string) string { return filepath.Join(r.LogDir, execID) } -// Run streams a chat completion to stdout.log. The response is wrapped in -// stream-json envelopes line-by-line so downstream parsers (summary, -// changestats) read it the same way they read Claude output. -func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution) error { +// 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") } @@ -70,56 +77,95 @@ func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Executio 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: t.Agent.Instructions}) + // The agent tools are real here, so guide the model toward them with the + // same planning preamble the container runners use (unless skip_planning). + messages = append(messages, llm.Message{Role: "user", Content: buildAgentInstructions(t, true)}) temperature := t.Agent.Temperature if temperature == nil && r.DefaultTemperature > 0 { v := r.DefaultTemperature temperature = &v } - - req := llm.ChatRequest{ - Model: t.Agent.Model, - Messages: messages, - Temperature: temperature, - MaxTokens: t.Agent.MaxTokens, - } + tools := agentToolDefs() start := time.Now() - resp, err := r.Client.ChatStream(ctx, req, func(delta string) { - if delta == "" { - return + var totalIn, totalOut int + var lastModel, lastFinish string + blocked := false + + for turn := 0; turn < maxLocalToolTurns; turn++ { + resp, chatErr := r.Client.Chat(ctx, llm.ChatRequest{ + Model: t.Agent.Model, + Messages: messages, + Temperature: temperature, + MaxTokens: t.Agent.MaxTokens, + Tools: tools, + }) + 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) } - writeAssistantTextLine(stdout, delta) - }) - if err != nil { - writeResultLine(stdout, "error", err.Error(), 0, 0) - return fmt.Errorf("local runner: chat: %w", err) - } - elapsed := time.Since(start) - // Write one consolidated assistant envelope containing the full response. - // extractSummary and ParseChangestatFromOutput operate per-line, so a - // single envelope with the full text is what they expect to find. - if resp.Content != "" { - writeAssistantTextLine(stdout, 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 + } } - writeResultLine(stdout, "success", "", resp.PromptTokens, resp.OutputTokens) + elapsed := time.Since(start) e.CostUSD = 0 - e.TokensIn = int64(resp.PromptTokens) - e.TokensOut = int64(resp.OutputTokens) + e.TokensIn = int64(totalIn) + e.TokensOut = int64(totalOut) if r.Logger != nil { r.Logger.Info("local runner completed", "taskID", t.ID, - "model", resp.Model, - "tokens_in", resp.PromptTokens, - "tokens_out", resp.OutputTokens, - "finish_reason", resp.FinishReason, + "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} + } + + writeResultLine(stdout, "success", "", totalIn, totalOut) return nil } diff --git a/internal/executor/local_test.go b/internal/executor/local_test.go index d8ab678..2ae8380 100644 --- a/internal/executor/local_test.go +++ b/internal/executor/local_test.go @@ -3,7 +3,7 @@ package executor import ( "context" "encoding/json" - "fmt" + "errors" "io" "log/slog" "net/http" @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "github.com/google/uuid" @@ -19,60 +20,77 @@ import ( "github.com/thepeterstone/claudomator/internal/task" ) -// fakeOpenAIServer returns an httptest.Server that replies with a streaming -// chat completion containing the supplied content (split into chunks) plus a -// usage record. -func fakeOpenAIServer(t *testing.T, chunks []string, promptTok, outTok int) *httptest.Server { +// fakeTurn is one canned chat-completion response the fake server returns. +type fakeTurn struct { + content string + toolCalls []llm.ToolCall + promptTok int + outTok int +} + +// fakeChatServer replies to /chat/completions with the supplied turns in order, +// one per request (non-streaming JSON), so a tool-use loop can be driven +// deterministically. Requests beyond the list repeat the final turn. +func fakeChatServer(t *testing.T, turns []fakeTurn) *httptest.Server { t.Helper() - return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "text/event-stream") - flusher, _ := w.(http.Flusher) - for _, c := range chunks { - payload := map[string]any{ - "model": "fake", - "choices": []map[string]any{{"delta": map[string]string{"content": c}}}, - } - b, _ := json.Marshal(payload) - fmt.Fprintf(w, "data: %s\n\n", b) - if flusher != nil { - flusher.Flush() - } + var mu sync.Mutex + idx := 0 + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + mu.Lock() + turn := turns[idx] + if idx < len(turns)-1 { + idx++ + } + mu.Unlock() + + msg := map[string]any{"role": "assistant", "content": turn.content} + if len(turn.toolCalls) > 0 { + msg["tool_calls"] = turn.toolCalls } - final := map[string]any{ + resp := map[string]any{ "model": "fake", - "choices": []map[string]any{{"delta": map[string]string{}, "finish_reason": "stop"}}, - "usage": map[string]int{"prompt_tokens": promptTok, "completion_tokens": outTok}, + "choices": []map[string]any{{"message": msg, "finish_reason": "stop"}}, + "usage": map[string]int{"prompt_tokens": turn.promptTok, "completion_tokens": turn.outTok}, } - fb, _ := json.Marshal(final) - fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", fb) + _ = json.NewEncoder(w).Encode(resp) })) } -func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { - srv := fakeOpenAIServer(t, - []string{"## Summary\n", "All ", "good."}, - 11, 22, - ) - defer srv.Close() +func toolCall(id, name, args string) llm.ToolCall { + return llm.ToolCall{ID: id, Type: "function", Function: llm.ToolCallFunction{Name: name, Arguments: args}} +} - logRoot := t.TempDir() - r := &LocalRunner{ +func newLocalRunner(t *testing.T, srv *httptest.Server) *LocalRunner { + t.Helper() + return &LocalRunner{ Client: &llm.Client{Endpoint: srv.URL + "/v1", Model: "fake"}, Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - LogDir: logRoot, + LogDir: t.TempDir(), } - tt := &task.Task{ +} + +func localTask() *task.Task { + return &task.Task{ ID: "task-1", Name: "test", Agent: task.AgentConfig{ Type: "local", Model: "fake", Instructions: "Do a thing.", + SkipPlanning: true, // keep the preamble out of these assertions }, } +} + +func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { + srv := fakeChatServer(t, []fakeTurn{{content: "## Summary\nAll good.", promptTok: 11, outTok: 22}}) + 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); err != nil { + if err := r.Run(context.Background(), tt, exec, newStoreChannel(&fakeChannelStore{}, tt.ID)); err != nil { t.Fatalf("Run: %v", err) } @@ -83,45 +101,108 @@ func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) { t.Errorf("tokens: want 11/22 got %d/%d", exec.TokensIn, exec.TokensOut) } - // Verify stdout.log contains stream-json envelopes that extractSummary can parse. stdoutPath := filepath.Join(r.ExecLogDir(exec.ID), "stdout.log") data, err := os.ReadFile(stdoutPath) if err != nil { t.Fatalf("read stdout: %v", err) } lines := strings.Split(strings.TrimSpace(string(data)), "\n") - if len(lines) < 4 { - t.Fatalf("expected at least 4 lines (3 deltas + 1 result), got %d:\n%s", len(lines), data) - } - for i, line := range lines[:3] { - var env struct { - Type string `json:"type"` - Message struct { - Content []struct { - Type string `json:"type"` - Text string `json:"text"` - } - } - } - if err := json.Unmarshal([]byte(line), &env); err != nil { - t.Fatalf("line %d not JSON: %v: %s", i, err, line) - } - if env.Type != "assistant" { - t.Errorf("line %d: want type=assistant, got %q", i, env.Type) - } + // One assistant envelope + one result line. + if len(lines) != 2 { + t.Fatalf("expected 2 lines (assistant + result), got %d:\n%s", len(lines), data) + } + var env struct { + Type string `json:"type"` + } + if err := json.Unmarshal([]byte(lines[0]), &env); err != nil || env.Type != "assistant" { + t.Fatalf("line 0 should be an assistant envelope: %v: %s", err, lines[0]) } - summary := extractSummary(stdoutPath) - if !strings.Contains(summary, "All good.") { + if summary := extractSummary(stdoutPath); !strings.Contains(summary, "All good.") { t.Errorf("extractSummary should find 'All good.', got %q", summary) } } +func TestLocalRunner_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}, + {toolCalls: []llm.ToolCall{toolCall("c2", "report_summary", `{"summary":"all done"}`)}, promptTok: 4, outTok: 2}, + {content: "finished", promptTok: 2, outTok: 1}, + }) + defer srv.Close() + + r := newLocalRunner(t, srv) + tt := localTask() + store := &fakeChannelStore{} + ch := newStoreChannel(store, tt.ID) + exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} + + if err := r.Run(context.Background(), tt, exec, ch); err != nil { + t.Fatalf("Run: %v", err) + } + + if len(store.createdTasks) != 1 || store.createdTasks[0].Name != "sub A" { + t.Errorf("expected one spawned subtask 'sub A', got %+v", store.createdTasks) + } + if store.createdTasks[0].ParentTaskID != tt.ID { + t.Errorf("subtask parent: want %q, got %q", tt.ID, store.createdTasks[0].ParentTaskID) + } + if sum, ok := ch.ReportedSummary(); !ok || sum != "all done" { + t.Errorf("expected reported summary 'all done', got %q (set=%v)", sum, ok) + } + // Tokens accumulate across all three turns. + if exec.TokensIn != 11 || exec.TokensOut != 6 { + t.Errorf("tokens should accumulate: want 11/6, got %d/%d", exec.TokensIn, exec.TokensOut) + } +} + +func TestLocalRunner_Run_RecordProgress(t *testing.T) { + srv := fakeChatServer(t, []fakeTurn{ + {toolCalls: []llm.ToolCall{toolCall("c1", "record_progress", `{"message":"halfway there"}`)}}, + {content: "done"}, + }) + defer srv.Close() + + store := &fakeChannelStore{} + tt := localTask() + exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} + if err := newLocalRunner(t, srv).Run(context.Background(), tt, exec, newStoreChannel(store, tt.ID)); err != nil { + t.Fatalf("Run: %v", err) + } + if len(store.createdEvents) != 1 { + t.Fatalf("expected 1 progress event, got %d", len(store.createdEvents)) + } + if !strings.Contains(string(store.createdEvents[0].Payload), "halfway there") { + t.Errorf("progress event payload: %s", store.createdEvents[0].Payload) + } +} + +func TestLocalRunner_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"}, + }) + defer srv.Close() + + tt := localTask() + exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID} + err := newLocalRunner(t, srv).Run(context.Background(), tt, exec, newStoreChannel(&fakeChannelStore{}, tt.ID)) + + var be *BlockedError + if !errors.As(err, &be) { + t.Fatalf("expected *BlockedError, got %v", err) + } + if !strings.Contains(be.QuestionJSON, "which branch?") { + t.Errorf("BlockedError question: %q", be.QuestionJSON) + } +} + func TestLocalRunner_Run_NoClient_Errors(t *testing.T) { r := &LocalRunner{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) + 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) } @@ -134,7 +215,7 @@ func TestLocalRunner_Run_EmptyInstructions_Errors(t *testing.T) { } tt := &task.Task{ID: "x", Agent: task.AgentConfig{}} exec := &storage.Execution{ID: "exec-x"} - err := r.Run(context.Background(), tt, exec) + err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)) if err == nil || !strings.Contains(err.Error(), "empty instructions") { t.Errorf("expected empty-instructions error, got %v", err) } diff --git a/internal/executor/localtools.go b/internal/executor/localtools.go new file mode 100644 index 0000000..48b862f --- /dev/null +++ b/internal/executor/localtools.go @@ -0,0 +1,135 @@ +package executor + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "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"}, + }, + }}, + } +} + +// dispatchAgentTool invokes one model-requested tool against the AgentChannel. +// 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, 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 + + default: + return "", false, fmt.Errorf("unknown tool %q", name) + } +} diff --git a/internal/executor/preamble.go b/internal/executor/preamble.go index b949986..77fae3c 100644 --- a/internal/executor/preamble.go +++ b/internal/executor/preamble.go @@ -2,61 +2,48 @@ package executor const planningPreamble = `## Runtime Environment -You are running as a background agent inside Claudomator. You cannot interact -with the user directly. However, if you need a decision or clarification: - -**To ask the user a question and pause:** -1. Write a JSON object to the path in $CLAUDOMATOR_QUESTION_FILE: - {"text": "Your question here?", "options": ["option A", "option B"]} - (options is optional — omit it for free-text answers) -2. Exit immediately. Do not wait. The task will be resumed with the user's answer - as the next message in this conversation. - -Only use this when you genuinely need user input to proceed. The text MUST be a -real question ending with "?". Do NOT write completion reports or status updates -here — use $CLAUDOMATOR_SUMMARY_FILE for those. Prefer making a reasonable -decision and noting it in your summary rather than asking. +You are running as a background agent inside Claudomator. You cannot chat with +the user directly, but you have these tools to coordinate: + +- **ask_user** — ask the user a question when you genuinely need a decision to + proceed. After calling it, end your turn immediately; the task pauses and is + resumed with the user's answer. Prefer making a reasonable decision and noting + it via report_summary over asking. +- **report_summary** — record a 2-5 sentence summary of what you did. Call this + before you finish so the user knows the outcome. +- **spawn_subtask** — create a child task to be run separately. +- **record_progress** — leave a short progress note in the task timeline. --- ## Planning Step (do this first) -Before doing any implementation work: - -1. Estimate: will this task take more than 3 minutes of implementation effort? +Before doing any implementation work, estimate whether the task will take more +than ~3 minutes of effort. -2. If YES — break it down: - - Create 3–7 discrete subtasks by POSTing to $CLAUDOMATOR_API_URL/api/tasks - - Each subtask POST body should be JSON with: name, agent.instructions, agent.project_dir (copy from $CLAUDOMATOR_PROJECT_DIR), agent.model, agent.allowed_tools, and agent.skip_planning set to true - - Set parent_task_id to $CLAUDOMATOR_TASK_ID in each POST body - - After creating all subtasks, output a brief summary and STOP. Do not implement anything. - - You can also specify agent.type (either "claude" or "gemini") to choose the agent for subtasks. - -3. If NO — proceed with the task instructions below. +- If YES — break it into 3-7 focused pieces with spawn_subtask, then call + report_summary describing the breakdown and STOP. Do not implement anything. +- If NO — proceed with the task instructions below. --- -## Git Discipline (mandatory when project_dir is set) +## Git Discipline (mandatory when working in a repository) -Every change you make to the working directory **must be committed before you finish**. -The sandbox is rejected if there are any uncommitted modifications. +Every change you make **must be committed before you finish** — the workspace is +rejected if there are uncommitted modifications. -- After completing work: run "git add -A && git commit -m 'concise description'" -- One commit is fine. Multiple focused commits are also fine. -- If you realise the task was already done and you made no changes, that is also fine — just exit cleanly without committing. -- Do not exit with uncommitted edits. -- **CRITICAL:** Run ALL git commands from your current directory — do NOT use absolute paths or "cd <project_path> && git ...". Your working directory IS the project. Using absolute paths bypasses the sandbox and breaks commit tracking. +- After completing work: run "git add -A && git commit -m 'concise description'". +- One commit is fine; multiple focused commits are also fine. +- If the task was already done and you made no changes, just exit cleanly. +- Run ALL git commands from your current directory — do NOT use absolute paths + or "cd <path> && git ...". Your working directory IS the project. --- -## Final Summary (mandatory) - -Before exiting, write a brief summary paragraph (2–5 sentences) describing what you did -and the outcome. Write it to the path in $CLAUDOMATOR_SUMMARY_FILE: - - echo "Your summary here." > "$CLAUDOMATOR_SUMMARY_FILE" +## Before Finishing -This summary is displayed in the task UI so the user knows what happened. +Call report_summary with a brief paragraph describing what you did and the +outcome. This is shown in the task UI. --- ` diff --git a/internal/executor/preamble_test.go b/internal/executor/preamble_test.go index 5c31b4f..77a8fca 100644 --- a/internal/executor/preamble_test.go +++ b/internal/executor/preamble_test.go @@ -5,27 +5,37 @@ import ( "testing" ) -func TestPlanningPreamble_ContainsFinalSummarySection(t *testing.T) { - if !strings.Contains(planningPreamble, "## Final Summary (mandatory)") { - t.Error("planningPreamble missing '## Final Summary (mandatory)' heading") +func TestPlanningPreamble_ReferencesMCPTools(t *testing.T) { + for _, tool := range []string{"ask_user", "report_summary", "spawn_subtask", "record_progress"} { + if !strings.Contains(planningPreamble, tool) { + t.Errorf("planningPreamble should mention the %q tool", tool) + } } } -func TestPlanningPreamble_SummaryUsesFileEnvVar(t *testing.T) { - if !strings.Contains(planningPreamble, "CLAUDOMATOR_SUMMARY_FILE") { - t.Error("planningPreamble should instruct agent to write summary to $CLAUDOMATOR_SUMMARY_FILE") +func TestPlanningPreamble_NoFileEscapeHatches(t *testing.T) { + for _, marker := range []string{"CLAUDOMATOR_SUMMARY_FILE", "CLAUDOMATOR_QUESTION_FILE"} { + if strings.Contains(planningPreamble, marker) { + t.Errorf("planningPreamble should no longer reference %s", marker) + } } } -func TestPlanningPreamble_SummaryInstructsEchoToFile(t *testing.T) { - if !strings.Contains(planningPreamble, `"$CLAUDOMATOR_SUMMARY_FILE"`) { - t.Error("planningPreamble should show example of writing to $CLAUDOMATOR_SUMMARY_FILE via echo") +func TestPlanningPreamble_AskUserEndsTurn(t *testing.T) { + if !strings.Contains(planningPreamble, "end your turn") { + t.Error("planningPreamble should instruct the agent to end its turn after ask_user") } } func TestPlanningPreamble_GitDiscipline_ForbidsAbsolutePaths(t *testing.T) { - // Agents must not bypass the sandbox by using absolute project paths in git commands. if !strings.Contains(planningPreamble, "do NOT use absolute paths") { t.Error("planningPreamble should warn agents not to use absolute paths in git commands") } } + +func TestWithPlanningPreamble_PrependsToInstructions(t *testing.T) { + got := withPlanningPreamble("DO THE THING") + if !strings.HasPrefix(got, planningPreamble) || !strings.HasSuffix(got, "DO THE THING") { + t.Error("withPlanningPreamble should prepend the preamble to the instructions") + } +} |