1 files changed, 159 insertions, 0 deletions
diff --git a/internal/executor/channel.go b/internal/executor/channel.go
new file mode 100644
index 0000000..8605ffa
--- /dev/null
+++ b/internal/executor/channel.go
@@ -0,0 +1,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,
+	})
+}
+\ No newline at end of file