1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
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,
})
}
|
