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 } func (c *storeChannel) ReportVerdict(_ context.Context, approved bool, reasoning string) error { payload, _ := json.Marshal(struct { Approved bool `json:"approved"` Reasoning string `json:"reasoning"` }{Approved: approved, Reasoning: reasoning}) return c.store.CreateEvent(&event.Event{ TaskID: c.taskID, Kind: event.KindVerdictReported, Actor: event.ActorAgent, Payload: payload, }) }