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
|
package executor
import (
"context"
"encoding/json"
"sync"
"time"
"github.com/thepeterstone/claudomator/internal/agentchannel"
"github.com/thepeterstone/claudomator/internal/event"
"github.com/thepeterstone/claudomator/internal/task"
"github.com/google/uuid"
)
// AgentChannel, SubtaskSpec, BlockedError, and ErrAgentBlocked used to be
// defined directly in this package. Phase 1 of the multi-provider refactor
// moved them to internal/agentchannel (a small leaf package with no
// executor dependency) so that internal/agentloop — the new provider-neutral
// tool-use loop — can construct/inspect them without importing
// internal/executor. That matters because internal/executor now imports
// internal/agentloop (via nativerunner.go, which wires an agentloop.Loop up
// to a Runner); if agentloop imported executor too, that would be a direct
// import cycle. Re-exporting via type aliases (and a var assignment for the
// sentinel error) here means every existing reference to
// executor.AgentChannel / executor.SubtaskSpec / executor.BlockedError /
// executor.ErrAgentBlocked keeps compiling and behaving identically — an
// alias is the same type, not a copy.
type AgentChannel = agentchannel.AgentChannel
type SubtaskSpec = agentchannel.SubtaskSpec
type BlockedError = agentchannel.BlockedError
var ErrAgentBlocked = agentchannel.ErrAgentBlocked
// 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,
})
}
|
