summaryrefslogtreecommitdiff
path: root/internal/executor
diff options
context:
space:
mode:
Diffstat (limited to 'internal/executor')
-rw-r--r--internal/executor/agentmcp.go160
-rw-r--r--internal/executor/agentmcp_test.go218
-rw-r--r--internal/executor/channel.go59
-rw-r--r--internal/executor/channel_test.go42
-rw-r--r--internal/executor/claude_test.go8
-rw-r--r--internal/executor/container_test.go12
-rw-r--r--internal/executor/executor.go12
-rw-r--r--internal/executor/gemini_test.go14
-rw-r--r--internal/executor/local_test.go6
9 files changed, 475 insertions, 56 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
index 76df94f..541694b 100644
--- a/internal/executor/channel.go
+++ b/internal/executor/channel.go
@@ -4,10 +4,10 @@ import (
"context"
"encoding/json"
"errors"
+ "sync"
"time"
"github.com/thepeterstone/claudomator/internal/event"
- "github.com/thepeterstone/claudomator/internal/storage"
"github.com/thepeterstone/claudomator/internal/task"
"github.com/google/uuid"
)
@@ -50,37 +50,64 @@ type channelStore interface {
CreateEvent(e *event.Event) error
}
-// storeChannel is the default AgentChannel backed by storage. Summary reports
-// are buffered onto the execution record so the pool persists them once (with
-// its extract/synthesize fallbacks); other signals are written immediately.
+// 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
- exec *storage.Execution
+
+ mu sync.Mutex
+ summary string
+ summarySet bool
+ question string
+ blocked bool
}
var _ AgentChannel = (*storeChannel)(nil)
-func newStoreChannel(store channelStore, taskID string, exec *storage.Execution) *storeChannel {
- return &storeChannel{store: store, taskID: taskID, exec: exec}
+func newStoreChannel(store channelStore, taskID string) *storeChannel {
+ return &storeChannel{store: store, taskID: taskID}
}
-// AskUser on the default (file-transport) channel cannot deliver an answer
-// in-session, so it always reports the run as blocked. Question persistence is
-// owned by the pool's BlockedError handling.
-func (c *storeChannel) AskUser(_ context.Context, _ string) (string, error) {
+// 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 onto the execution record. The pool persists
-// it (preferring this value over its extract/synthesize fallbacks).
+// 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 {
- if c.exec != nil {
- c.exec.Summary = summary
- }
+ 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{
diff --git a/internal/executor/channel_test.go b/internal/executor/channel_test.go
index 822dd35..250e8eb 100644
--- a/internal/executor/channel_test.go
+++ b/internal/executor/channel_test.go
@@ -6,7 +6,6 @@ import (
"testing"
"github.com/thepeterstone/claudomator/internal/event"
- "github.com/thepeterstone/claudomator/internal/storage"
"github.com/thepeterstone/claudomator/internal/task"
)
@@ -29,8 +28,8 @@ func (f *fakeChannelStore) CreateEvent(e *event.Event) error {
return nil
}
-func TestStoreChannel_AskUser_ReturnsBlocked(t *testing.T) {
- ch := newStoreChannel(&fakeChannelStore{}, "task-1", &storage.Execution{})
+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)
@@ -38,29 +37,36 @@ func TestStoreChannel_AskUser_ReturnsBlocked(t *testing.T) {
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_ReportSummary_BuffersOntoExec(t *testing.T) {
- e := &storage.Execution{}
- ch := newStoreChannel(&fakeChannelStore{}, "task-1", e)
- if err := ch.ReportSummary(context.Background(), "did the thing"); err != nil {
- t.Fatalf("ReportSummary: %v", err)
- }
- if e.Summary != "did the thing" {
- t.Errorf("expected exec.Summary set, got %q", e.Summary)
+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_NilExec(t *testing.T) {
- ch := newStoreChannel(&fakeChannelStore{}, "task-1", nil)
- if err := ch.ReportSummary(context.Background(), "x"); err != nil {
- t.Errorf("expected nil err with nil exec, got %v", err)
+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", &storage.Execution{})
+ ch := newStoreChannel(store, "parent-1")
id, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{
Name: "child",
Instructions: "do it",
@@ -93,7 +99,7 @@ func TestStoreChannel_SpawnSubtask_CreatesChildWithParent(t *testing.T) {
func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) {
store := &fakeChannelStore{createTaskErr: errors.New("boom")}
- ch := newStoreChannel(store, "parent-1", &storage.Execution{})
+ ch := newStoreChannel(store, "parent-1")
if _, err := ch.SpawnSubtask(context.Background(), SubtaskSpec{Name: "x"}); err == nil {
t.Error("expected error from CreateTask")
}
@@ -101,7 +107,7 @@ func TestStoreChannel_SpawnSubtask_PropagatesError(t *testing.T) {
func TestStoreChannel_RecordProgress_EmitsAgentMessage(t *testing.T) {
store := &fakeChannelStore{}
- ch := newStoreChannel(store, "task-1", &storage.Execution{})
+ ch := newStoreChannel(store, "task-1")
if err := ch.RecordProgress(context.Background(), "halfway done"); err != nil {
t.Fatalf("RecordProgress: %v", err)
}
diff --git a/internal/executor/claude_test.go b/internal/executor/claude_test.go
index 0e76260..414f6cf 100644
--- a/internal/executor/claude_test.go
+++ b/internal/executor/claude_test.go
@@ -259,7 +259,7 @@ func TestClaudeRunner_Run_ResumeSetsSessionIDFromResumeSession(t *testing.T) {
}
// Run completes successfully (binary is "true").
- _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
+ _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID))
// 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
@@ -284,7 +284,7 @@ func TestClaudeRunner_Run_InaccessibleWorkingDir_ReturnsError(t *testing.T) {
}
exec := &storage.Execution{ID: "test-exec"}
- err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
+ err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID))
if err == nil {
t.Fatal("expected error for inaccessible working_dir, got nil")
@@ -732,7 +732,7 @@ func TestClaudeRunner_Run_ResumeUsesStoredSandboxDir(t *testing.T) {
SandboxDir: sandboxDir,
}
- _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
+ _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID))
got, err := os.ReadFile(cwdFile)
if err != nil {
@@ -778,7 +778,7 @@ func TestClaudeRunner_Run_StaleSandboxDir_ClonesAfresh(t *testing.T) {
SandboxDir: staleSandbox,
}
- if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil {
t.Fatalf("Run with stale sandbox: %v", err)
}
diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go
index 5ee3a3c..9cd80dc 100644
--- a/internal/executor/container_test.go
+++ b/internal/executor/container_test.go
@@ -139,7 +139,7 @@ func TestContainerRunner_Run_PreservesWorkspaceOnFailure(t *testing.T) {
}
exec := &storage.Execution{ID: "test-exec", TaskID: "test-task"}
- err := runner.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
+ err := runner.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID))
if err == nil {
t.Fatal("expected error due to mocked docker failure")
}
@@ -378,7 +378,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, newStoreChannel(nil, tk.ID, 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 +418,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, newStoreChannel(nil, tk.ID, 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 +504,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, newStoreChannel(nil, tk.ID, 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 +550,7 @@ func TestContainerRunner_ClonesStoryBranch(t *testing.T) {
}
e := &storage.Execution{ID: "exec-1", TaskID: "story-branch-test"}
- runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, 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 +597,7 @@ func TestContainerRunner_ClonesDefaultBranchWhenNoBranchName(t *testing.T) {
}
e := &storage.Execution{ID: "exec-2", TaskID: "no-branch-test"}
- runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
+ runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID))
os.RemoveAll(e.SandboxDir)
for _, a := range cloneArgs {
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 76e67b8..6d6c528 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -356,8 +356,12 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex
}
}
- err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, 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)
@@ -1077,8 +1081,12 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
}
// Run the task.
- err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, 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/gemini_test.go b/internal/executor/gemini_test.go
index c8f7422..a906f2c 100644
--- a/internal/executor/gemini_test.go
+++ b/internal/executor/gemini_test.go
@@ -125,7 +125,7 @@ func TestGeminiRunner_Run_InaccessibleProjectDir_ReturnsError(t *testing.T) {
}
exec := &storage.Execution{ID: "test-exec"}
- err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
+ err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID))
if err == nil {
t.Fatal("expected error for inaccessible project_dir, got nil")
@@ -213,7 +213,7 @@ func TestGeminiRunner_Run_ProjectDir_RunsInSandbox(t *testing.T) {
}
e := &storage.Execution{ID: "sandbox-exec", TaskID: "task-1"}
- if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil {
t.Fatalf("Run: %v", err)
}
@@ -261,7 +261,7 @@ fi
}
e := &storage.Execution{ID: "blocked-gemini-exec", TaskID: "task-1"}
- err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
+ err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID))
var blocked *BlockedError
if !errors.As(err, &blocked) {
@@ -301,7 +301,7 @@ func TestGeminiRunner_Run_ExecError_PreservesSandbox(t *testing.T) {
}
e := &storage.Execution{ID: "err-gemini-exec", TaskID: "task-1"}
- err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
+ err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID))
if err == nil {
t.Fatal("expected error from failing gemini exit")
}
@@ -352,7 +352,7 @@ func TestGeminiRunner_Run_ResumeUsesStoredSandboxDir(t *testing.T) {
SandboxDir: sandboxDir,
}
- if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil {
t.Fatalf("Run with preserved sandbox: %v", err)
}
@@ -400,7 +400,7 @@ func TestGeminiRunner_Run_StaleSandboxDir_ClonesAfresh(t *testing.T) {
SandboxDir: staleSandbox,
}
- if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil {
t.Fatalf("Run with stale sandbox: %v", err)
}
@@ -438,7 +438,7 @@ func TestGeminiRunner_Run_NoProjectDir_SkipsSandbox(t *testing.T) {
}
e := &storage.Execution{ID: "no-pd-gemini", TaskID: "task-nopd"}
- if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID)); err != nil {
t.Fatalf("Run without project_dir: %v", err)
}
if e.SandboxDir != "" {
diff --git a/internal/executor/local_test.go b/internal/executor/local_test.go
index 8aa8791..ffe87f9 100644
--- a/internal/executor/local_test.go
+++ b/internal/executor/local_test.go
@@ -72,7 +72,7 @@ func TestLocalRunner_Run_WritesStreamJSON(t *testing.T) {
}
exec := &storage.Execution{ID: uuid.New().String(), TaskID: tt.ID}
- if err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec)); err != nil {
+ if err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID)); err != nil {
t.Fatalf("Run: %v", err)
}
@@ -121,7 +121,7 @@ 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, newStoreChannel(nil, tt.ID, 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 +134,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, newStoreChannel(nil, tt.ID, 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)
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 20:18:07 +0000