summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--internal/api/server_test.go2
-rw-r--r--internal/executor/channel.go117
-rw-r--r--internal/executor/channel_test.go118
-rw-r--r--internal/executor/claude.go6
-rw-r--r--internal/executor/claude_test.go8
-rw-r--r--internal/executor/container.go12
-rw-r--r--internal/executor/container_test.go12
-rw-r--r--internal/executor/executor.go12
-rw-r--r--internal/executor/executor_test.go8
-rw-r--r--internal/executor/gemini.go6
-rw-r--r--internal/executor/gemini_test.go14
-rw-r--r--internal/executor/local.go2
-rw-r--r--internal/executor/local_test.go6
13 files changed, 282 insertions, 41 deletions
diff --git a/internal/api/server_test.go b/internal/api/server_test.go
index 2530d55..f902495 100644
--- a/internal/api/server_test.go
+++ b/internal/api/server_test.go
@@ -116,7 +116,7 @@ func (m *mockRunner) ExecLogDir(execID string) string {
return filepath.Join(m.logDir, execID)
}
-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, _ executor.AgentChannel) error {
if e.ID == "" {
e.ID = uuid.New().String()
}
diff --git a/internal/executor/channel.go b/internal/executor/channel.go
new file mode 100644
index 0000000..76df94f
--- /dev/null
+++ b/internal/executor/channel.go
@@ -0,0 +1,117 @@
+package executor
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "time"
+
+ "github.com/thepeterstone/claudomator/internal/event"
+ "github.com/thepeterstone/claudomator/internal/storage"
+ "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
+}
+
+// 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.
+type storeChannel struct {
+ store channelStore
+ taskID string
+ exec *storage.Execution
+}
+
+var _ AgentChannel = (*storeChannel)(nil)
+
+func newStoreChannel(store channelStore, taskID string, exec *storage.Execution) *storeChannel {
+ return &storeChannel{store: store, taskID: taskID, exec: exec}
+}
+
+// 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) {
+ return "", ErrAgentBlocked
+}
+
+// ReportSummary buffers the summary onto the execution record. The pool persists
+// it (preferring this value over its extract/synthesize fallbacks).
+func (c *storeChannel) ReportSummary(_ context.Context, summary string) error {
+ if c.exec != nil {
+ c.exec.Summary = summary
+ }
+ return nil
+}
+
+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..822dd35
--- /dev/null
+++ b/internal/executor/channel_test.go
@@ -0,0 +1,118 @@
+package executor
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/thepeterstone/claudomator/internal/event"
+ "github.com/thepeterstone/claudomator/internal/storage"
+ "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_ReturnsBlocked(t *testing.T) {
+ ch := newStoreChannel(&fakeChannelStore{}, "task-1", &storage.Execution{})
+ 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)
+ }
+}
+
+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_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_SpawnSubtask_CreatesChildWithParent(t *testing.T) {
+ store := &fakeChannelStore{}
+ ch := newStoreChannel(store, "parent-1", &storage.Execution{})
+ 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", &storage.Execution{})
+ 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", &storage.Execution{})
+ 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
index 3c87f26..0123754 100644
--- a/internal/executor/claude.go
+++ b/internal/executor/claude.go
@@ -51,7 +51,7 @@ func (r *ClaudeRunner) binaryPath() string {
// 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 {
+func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error {
projectDir := t.Agent.ProjectDir
// Validate project_dir exists when set.
@@ -164,7 +164,7 @@ func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi
// 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)
+ _ = ch.ReportSummary(ctx, 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.
@@ -177,7 +177,7 @@ func (r *ClaudeRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi
summaryFile := filepath.Join(logDir, "summary.txt")
if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil {
os.Remove(summaryFile) // consumed
- e.Summary = strings.TrimSpace(string(summaryData))
+ _ = ch.ReportSummary(ctx, strings.TrimSpace(string(summaryData)))
}
// Merge sandbox back to project_dir and clean up.
diff --git a/internal/executor/claude_test.go b/internal/executor/claude_test.go
index c01e160..0e76260 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)
+ _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, 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
@@ -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)
+ err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
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)
+ _ = r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
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); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
t.Fatalf("Run with stale sandbox: %v", err)
}
diff --git a/internal/executor/container.go b/internal/executor/container.go
index 61ac29c..4269ef1 100644
--- a/internal/executor/container.go
+++ b/internal/executor/container.go
@@ -100,7 +100,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 == "" {
@@ -227,7 +227,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()
@@ -241,7 +241,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 {
@@ -257,7 +257,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
@@ -372,7 +372,7 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto
questionJSON := strings.TrimSpace(string(data))
if isCompletionReport(questionJSON) {
r.Logger.Info("treating question file as completion report", "taskID", e.TaskID)
- e.Summary = extractQuestionText(questionJSON)
+ _ = ch.ReportSummary(ctx, extractQuestionText(questionJSON))
} else {
if e.SessionID == "" {
r.Logger.Warn("missing session ID; resume will start fresh", "taskID", e.TaskID)
@@ -389,7 +389,7 @@ func (r *ContainerRunner) runContainer(ctx context.Context, t *task.Task, e *sto
summaryFile := filepath.Join(logDir, "summary.txt")
if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil {
os.Remove(summaryFile) // consumed
- e.Summary = strings.TrimSpace(string(summaryData))
+ _ = ch.ReportSummary(ctx, strings.TrimSpace(string(summaryData)))
}
// 5. Post-execution: push changes if successful
diff --git a/internal/executor/container_test.go b/internal/executor/container_test.go
index f0b2a3a..5ee3a3c 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)
+ err := runner.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
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)
+ err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
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)
+ err := runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
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)
+ runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
// 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)
+ runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
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)
+ runner.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
os.RemoveAll(e.SandboxDir)
for _, a := range cloneArgs {
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 09169bd..76e67b8 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.
@@ -352,7 +356,7 @@ func (p *Pool) executeResume(ctx context.Context, t *task.Task, exec *storage.Ex
}
}
- err = runner.Run(ctx, t, exec)
+ err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, exec))
exec.EndTime = time.Now().UTC()
p.decActiveAgent(agentType, &cleaned)
@@ -1073,7 +1077,7 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) {
}
// Run the task.
- err = runner.Run(ctx, t, exec)
+ err = runner.Run(ctx, t, exec, newStoreChannel(p.store, t.ID, exec))
exec.EndTime = time.Now().UTC()
p.decActiveAgent(agentType, &cleaned)
diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go
index 9214872..d98efbf 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
@@ -391,11 +392,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 +1195,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
index 3abec05..d229361 100644
--- a/internal/executor/gemini.go
+++ b/internal/executor/gemini.go
@@ -49,7 +49,7 @@ func (r *GeminiRunner) binaryPath() string {
// 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 {
+func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, ch AgentChannel) error {
projectDir := t.Agent.ProjectDir
if projectDir != "" {
@@ -136,7 +136,7 @@ func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi
questionJSON := strings.TrimSpace(string(data))
if isCompletionReport(questionJSON) {
r.Logger.Info("treating question file as completion report", "taskID", e.TaskID)
- e.Summary = extractQuestionText(questionJSON)
+ _ = ch.ReportSummary(ctx, 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}
@@ -147,7 +147,7 @@ func (r *GeminiRunner) Run(ctx context.Context, t *task.Task, e *storage.Executi
summaryFile := filepath.Join(logDir, "summary.txt")
if summaryData, readErr := os.ReadFile(summaryFile); readErr == nil {
os.Remove(summaryFile)
- e.Summary = strings.TrimSpace(string(summaryData))
+ _ = ch.ReportSummary(ctx, strings.TrimSpace(string(summaryData)))
}
// Merge sandbox back to project_dir and clean up.
diff --git a/internal/executor/gemini_test.go b/internal/executor/gemini_test.go
index cd11ebc..c8f7422 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)
+ err := r.Run(context.Background(), tk, exec, newStoreChannel(nil, tk.ID, exec))
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); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); 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)
+ err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
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)
+ err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e))
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); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); 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); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); 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); err != nil {
+ if err := r.Run(context.Background(), tk, e, newStoreChannel(nil, tk.ID, e)); err != nil {
t.Fatalf("Run without project_dir: %v", err)
}
if e.SandboxDir != "" {
diff --git a/internal/executor/local.go b/internal/executor/local.go
index 5d874c6..e3c9c2c 100644
--- a/internal/executor/local.go
+++ b/internal/executor/local.go
@@ -40,7 +40,7 @@ func (r *LocalRunner) ExecLogDir(execID string) string {
// 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 {
+func (r *LocalRunner) Run(ctx context.Context, t *task.Task, e *storage.Execution, _ AgentChannel) error {
if r.Client == nil {
return fmt.Errorf("local runner: no LLM client configured")
}
diff --git a/internal/executor/local_test.go b/internal/executor/local_test.go
index d8ab678..8aa8791 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); err != nil {
+ if err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec)); 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)
+ err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec))
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)
+ err := r.Run(context.Background(), tt, exec, newStoreChannel(nil, tt.ID, exec))
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 14:15:16 +0000