diff options
Diffstat (limited to 'internal/executor')
| -rw-r--r-- | internal/executor/channel.go | 15 | ||||
| -rw-r--r-- | internal/executor/container.go | 63 | ||||
| -rw-r--r-- | internal/executor/container_test.go | 58 |
3 files changed, 128 insertions, 8 deletions
diff --git a/internal/executor/channel.go b/internal/executor/channel.go index 541694b..8605ffa 100644 --- a/internal/executor/channel.go +++ b/internal/executor/channel.go @@ -50,6 +50,21 @@ type channelStore interface { 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 diff --git a/internal/executor/container.go b/internal/executor/container.go index 4269ef1..f0f728c 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 } @@ -307,8 +311,24 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto return fmt.Errorf("writing instructions: %w", err) } + // Per-task MCP back-channel: mint a scoped token bound to this run's channel + // and write an mcp-config pointing the agent at the host agent MCP server. + mcpEnabled := false + if r.Registry != nil && t.Agent.Type != "gemini" { + 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 err := writeMCPConfig(workspace, mcpURL, token); err != nil { + return fmt.Errorf("writing mcp config: %w", err) + } + mcpEnabled = true + } + 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...) @@ -365,7 +385,17 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto e.SessionID = sessionID } - // Check whether the agent left a question before exiting. + // 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) + } + return &BlockedError{QuestionJSON: q, SessionID: e.SessionID, SandboxDir: workspace} + } + + // Check whether the agent left a question before exiting (file fallback for + // in-flight tasks started on the pre-MCP wire). questionFile := filepath.Join(logDir, "question.json") if data, readErr := os.ReadFile(questionFile); readErr == nil { os.Remove(questionFile) // consumed @@ -467,7 +497,30 @@ 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" + +// 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) +} + +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. @@ -491,6 +544,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()} @@ -501,6 +557,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 9cd80dc..86e95e2 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" @@ -53,13 +54,44 @@ func TestContainerRunner_BuildDockerArgs(t *testing.T) { } } +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 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 +105,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 +113,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 +147,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) } |