path: root/internal/executor/channel.go

package executor

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"sync"
	"time"

	"github.com/thepeterstone/claudomator/internal/agentchannel"
	"github.com/thepeterstone/claudomator/internal/event"
	"github.com/thepeterstone/claudomator/internal/role"
	"github.com/thepeterstone/claudomator/internal/storage"
	"github.com/thepeterstone/claudomator/internal/story"
	"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

// EpicProposal is the Phase 7c sibling of SubtaskSpec above: an alias so
// existing code that refers to executor.EpicProposal (agentmcp.go, tests)
// compiles against the same type agentchannel.AgentChannel.ProposeEpic uses.
type EpicProposal = agentchannel.EpicProposal

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
	// CreateEpic, GetEpicByName, GetStory, and UpdateStory back
	// storeChannel.ProposeEpic (Phase 7c): matching an existing epic by exact
	// name (or creating a new one), then attaching the given stories to it.
	CreateEpic(e *story.Epic) error
	GetEpicByName(name string) (*story.Epic, error)
	GetStory(id string) (*story.Story, error)
	UpdateStory(st *story.Story) error
	// CreateRoleConfig backs storeChannel.ProposeRoleConfig (Phase 8): it
	// inserts a new "draft"-status role_configs row for a role, auto-
	// assigning the next version number, exactly the way
	// api.handleCreateRoleVersion (the human-facing
	// POST /api/roles/{role}/versions endpoint) already uses it.
	CreateRoleConfig(roleName, configJSON, proposedBy string) (*storage.RoleConfigRow, 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()
	agent := task.AgentConfig{
		Instructions: spec.Instructions,
		MaxBudgetUSD: spec.MaxBudgetUSD,
	}
	if spec.Role != "" {
		// Role-typed subtask: leave Type/Model empty so
		// executor.Pool.execute() resolves tier 0 of the role's escalation
		// ladder on dispatch, exactly like any other role-typed task.
		agent.Role = spec.Role
	} else {
		// Backward-compatible default: every pre-Phase-6 caller of
		// SpawnSubtask gets exactly this unchanged behavior.
		agent.Type = "claude"
		agent.Model = spec.Model
	}
	child := &task.Task{
		ID:           uuid.NewString(),
		Name:         spec.Name,
		ParentTaskID: c.taskID,
		Agent:        agent,
		Priority:     task.PriorityNormal,
		DependsOn:    spec.DependsOn,
		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,
	})
}

// ProposeEpic groups spec.StoryIDs under a new or existing epic, matched by
// exact spec.Name — the simplest reasonable de-dup; a fuzzy-matching pass is
// left to a later phase if it turns out to matter. Story IDs that don't
// resolve are skipped (not fatal to the call): one bad ID shouldn't prevent
// grouping the rest. The event is attached to the epic's own ID, matching
// the story orchestrator's convention (KindEvalVerdict/KindArbitrationDecided)
// of attaching planning-layer ceremony events to the entity they're about
// rather than to a task ID — events.task_id has no enforced FK, so this is
// exactly the tolerance already built in for that.
func (c *storeChannel) ProposeEpic(_ context.Context, spec agentchannel.EpicProposal) (string, error) {
	epic, err := c.store.GetEpicByName(spec.Name)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			return "", err
		}
		epic = &story.Epic{
			ID:              uuid.NewString(),
			Name:            spec.Name,
			Description:     spec.Description,
			DiscoverySource: "agent",
		}
		if err := c.store.CreateEpic(epic); err != nil {
			return "", err
		}
	}

	grouped := make([]string, 0, len(spec.StoryIDs))
	for _, sid := range spec.StoryIDs {
		st, err := c.store.GetStory(sid)
		if err != nil {
			continue // unresolved story ID: skip, don't fail the whole call
		}
		st.EpicID = epic.ID
		if err := c.store.UpdateStory(st); err != nil {
			continue
		}
		grouped = append(grouped, sid)
	}

	payload, _ := json.Marshal(struct {
		EpicID   string   `json:"epic_id"`
		Name     string   `json:"name"`
		StoryIDs []string `json:"story_ids"`
	}{EpicID: epic.ID, Name: epic.Name, StoryIDs: grouped})
	if err := c.store.CreateEvent(&event.Event{
		TaskID:  epic.ID,
		Kind:    event.KindEpicProposed,
		Actor:   event.ActorAgent,
		Payload: payload,
	}); err != nil {
		return epic.ID, err
	}
	return epic.ID, nil
}

// ProposeRoleConfig creates a new draft role_configs row for config.Role,
// proposed_by "retro" (the only role this phase wires the tool up for,
// though nothing else about the mechanism is retro-specific). It never reads
// or touches whatever version is currently active for that role — draft
// creation and activation are deliberately separate paths (see
// storage.CreateRoleConfig / storage.ActivateRoleConfigVersion), the latter
// staying a human action via POST /api/roles/{role}/activate.
//
// Records event.KindRoleConfigProposed on the *calling task's own ID*
// (c.taskID, not a role or story ID — roles have no first-class row/ID of
// their own the way epics/stories do), payload {role, version}. This mirrors
// ProposeEpic's "one event per proposal call" convention but attaches to the
// task rather than the proposed entity: internal/scheduler.StoryOrchestrator
// reads a retro task's own event stream for these once the task completes,
// to assemble the aggregate KindRetroCaptured payload it attaches to the
// story's ID (see that package's finalizeRetro).
func (c *storeChannel) ProposeRoleConfig(_ context.Context, config role.RoleConfig) (int, error) {
	configJSON, err := json.Marshal(config)
	if err != nil {
		return 0, err
	}
	row, err := c.store.CreateRoleConfig(config.Role, string(configJSON), "retro")
	if err != nil {
		return 0, err
	}

	payload, _ := json.Marshal(struct {
		Role    string `json:"role"`
		Version int    `json:"version"`
	}{Role: row.Role, Version: row.Version})
	if err := c.store.CreateEvent(&event.Event{
		TaskID:  c.taskID,
		Kind:    event.KindRoleConfigProposed,
		Actor:   event.ActorAgent,
		Payload: payload,
	}); err != nil {
		return row.Version, err
	}
	return row.Version, nil
}