path: root/internal/scheduler/story_orchestrator.go

// This file implements Phase 7b's story orchestrator: a deterministic,
// poll-based watcher that drives a story.Story through
// Builder -> 4 parallel Evaluators -> Arbitration -> a single human
// accept-gate at the *story* level, exactly the way Scheduler (above, in
// scheduler.go) drives a role-typed task through its retry/escalation
// ladder. It lives in this package rather than internal/story for a
// structural reason, not just stylistic taste: internal/storage already
// imports internal/story (storage/story.go), so internal/story cannot
// import internal/storage back — and this orchestrator's Store interface
// needs storage.StoryFilter/*task.Task/etc. internal/scheduler already
// imports both internal/storage and internal/task (and has zero risk of
// internal/story importing scheduler back), so it's a clean home. It also
// matches the design note this phase's task description invited: "fold it
// into internal/scheduler as a sibling to the existing Scheduler".
//
// Mechanism choice: poll-based, not a handleRunResult hook. Every task this
// orchestrator cares about (the story's root Builder task, the 4 Evaluators,
// the Arbitration task) is a top-level task (ParentTaskID == ""), so per
// task.go's state machine the *only* way one of them would ever reach
// COMPLETED on its own is via a human/chatbot POST /api/tasks/{id}/accept
// (internal/api.acceptTask), READY -> COMPLETED — executor.Pool.handleRunResult
// never transitions a top-level task to COMPLETED directly (only READY, or
// BLOCKED if it has subtasks). That's exactly the gap this orchestrator
// closes itself (see autoAccept below): the whole point of the story-level
// ceremony is that a human/chatbot should only ever have to make *one*
// accept decision (the final POST /api/stories/{id}/accept, REVIEW_READY ->
// DONE) — not six (builder + 4 evaluators + arbitration) along the way. So
// on every tick, for the specific Builder/Evaluator/Arbitration tasks it
// already knows are wired into a story's pipeline, this orchestrator
// auto-accepts (READY -> COMPLETED) them itself, using the same
// state-machine-respecting write internal/api's acceptTask uses
// (Store.UpdateTaskState, which wraps storage.DB.UpdateTaskStateBy —
// validates task.ValidTransition and writes the state_change event
// atomically, not a raw/unchecked write). This is narrowly scoped: it only
// ever touches the root task a story actually tracks (st.RootTaskID) and
// that root's structurally-discovered evaluator/arbitration dependents
// (found via ensureEvaluators/ensureArbitration's own role-matching) — never
// a blanket "auto-accept every READY task" sweep. Polling (rather than a
// handleRunResult hook) is still the right mechanism regardless of who does
// the accepting: the transition happens via a *write this orchestrator
// itself performs* on a poll tick, which is inherently poll-driven, not an
// executor-package callback.
package scheduler

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/thepeterstone/claudomator/internal/event"
	"github.com/thepeterstone/claudomator/internal/storage"
	"github.com/thepeterstone/claudomator/internal/story"
	"github.com/thepeterstone/claudomator/internal/task"
)

// evaluatorRoles is the fixed fan-out of Evaluator roles spawned once a
// story's Builder (root_task_id) task completes. Order here is preserved
// wherever evaluator tasks are collected into a slice (e.g. building the
// Arbitration task's DependsOn), purely for determinism/testability — the
// orchestrator does not otherwise care about ordering.
var evaluatorRoles = []string{
	"evaluator_quality",
	"evaluator_security",
	"evaluator_correctness",
	"evaluator_performance",
}

// arbitrationRole is the role assigned to the single task spawned once all
// Evaluators for a story complete. It depends on all 4 Evaluator tasks and
// routes the story to REVIEW_READY or NEEDS_FIX based on a structured
// KindVerdictReported event (see finalizeArbitration/arbitrationVerdict).
const arbitrationRole = "planner"

// retroRole is the role assigned to the single task spawned once a story
// reaches DONE (see processRetro below) to reflect on the whole story and
// propose role_configs improvements — the final stage in the harness's
// self-improvement loop (Phase 8): retro produces draft role_configs
// versions, a human reviews and activates them via the existing
// POST /api/roles/{role}/activate, and future dispatches use the improved
// config. Just another role in the system, resolved via the same
// escalation-ladder machinery as builder/evaluator/planner — not a special
// case.
const retroRole = "retro"

// StoryStore is the subset of storage.DB methods StoryOrchestrator needs.
// Defining it as an interface (mirroring executor.Store/scheduler.Store)
// allows tests to supply an in-memory fake with no real SQLite database.
type StoryStore interface {
	ListStories(filter storage.StoryFilter) ([]*story.Story, error)
	UpdateStory(st *story.Story) error
	GetTask(id string) (*task.Task, error)
	ListDependents(taskID string) ([]*task.Task, error)
	CreateTask(t *task.Task) error
	UpdateTaskState(id string, newState task.State) error
	CreateEvent(e *event.Event) error
	// ListSubtasks, ListExecutions, GetActiveRoleConfig, and ListEvents back
	// the Phase 8 retro stage (processRetro/buildRetroInstructions/
	// finalizeRetro below): assembling a story's full task tree (subtasks +
	// dependents, mirroring GET /api/stories/{id}/task-tree's walk in
	// internal/api/stories.go), each task's cost/escalation history, the
	// active config for each role involved, and the story's own event
	// stream (evaluator verdicts, arbitration decision) into the retro
	// task's instructions, then reading the retro task's own event stream
	// back once it completes to see what it proposed.
	ListSubtasks(parentID string) ([]*task.Task, error)
	ListExecutions(taskID string) ([]*storage.Execution, error)
	GetActiveRoleConfig(roleName string) (*storage.RoleConfigRow, error)
	ListEvents(taskID string, sinceSeq int64) ([]*event.Event, error)
}

// StoryOrchestrator polls stories with a root_task_id set and drives them
// through the Builder -> Evaluators -> Arbitration -> REVIEW_READY ceremony,
// auto-accepting (READY -> COMPLETED) the Builder/Evaluator/Arbitration tasks
// along the way (see autoAccept) so that a human/chatbot never has to touch
// POST /api/tasks/{id}/accept for any of them. The final REVIEW_READY -> DONE
// transition is the *only* remaining human/chatbot action (see internal/api's
// POST /api/stories/{id}/accept), not something this type does itself.
type StoryOrchestrator struct {
	Store StoryStore
	// Pool reuses the same minimal interface Scheduler depends on
	// (Submit(ctx, *task.Task) error) — satisfied directly by
	// *executor.Pool, declared once in scheduler.go.
	Pool   Pool
	Logger *slog.Logger

	// handledVerdicts dedupes per-evaluator KindEvalVerdict emission within a
	// single running process, keyed by evaluator task ID. Without it, an
	// evaluator that completes while its siblings are still running would
	// get a fresh eval_verdict event on every poll tick until the last
	// sibling finishes. This mirrors Scheduler.handled exactly: an in-memory,
	// per-process guard that resets on restart. That's an accepted
	// simplification here for the same reason it is for Scheduler — this is
	// idempotent bookkeeping on an append-only observability stream, not
	// orchestration state; a restart can produce at most one duplicate
	// eval_verdict event per evaluator, never an infinite loop, because the
	// *structural* idempotency checks (ensureEvaluators/ensureArbitration
	// below, and the story.Status=="VALIDATING" gate in
	// finalizeArbitration) are what actually prevent duplicate task
	// creation and duplicate story-status transitions — the two things that
	// would matter if repeated forever.
	handledVerdicts map[string]bool

	// handledRetro dedupes KindRetroCaptured emission within a single
	// running process, keyed by retro task ID — the same in-memory,
	// per-process guard handledVerdicts is for eval_verdict, and for the
	// same reason: the structural idempotency check that actually matters
	// (processRetro's "does a retro-role task already depend on the
	// arbitration task" check, mirroring ensureEvaluators/ensureArbitration)
	// prevents ever spawning a second retro task; this only prevents
	// re-emitting the summary event every tick while the story sits DONE.
	handledRetro map[string]bool
}

// DefaultStoryPollInterval is used by Run when pollInterval <= 0.
const DefaultStoryPollInterval = 15 * time.Second

// Run polls all stories with a root_task_id set every pollInterval until ctx
// is cancelled.
func (o *StoryOrchestrator) Run(ctx context.Context, pollInterval time.Duration) {
	if pollInterval <= 0 {
		pollInterval = DefaultStoryPollInterval
	}
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			o.Tick(ctx)
		}
	}
}

// Tick runs a single poll pass over every story. Exported so tests can drive
// it directly without waiting on a ticker.
func (o *StoryOrchestrator) Tick(ctx context.Context) {
	stories, err := o.Store.ListStories(storage.StoryFilter{})
	if err != nil {
		o.logf("story orchestrator: list stories", "error", err)
		return
	}
	for _, st := range stories {
		if st.RootTaskID == "" {
			continue // no execution tree yet — nothing for this orchestrator to do
		}
		if st.Status == "CANCELLED" {
			continue // terminal; this orchestrator never revives a story from here
		}
		if st.Status == "DONE" {
			// The Builder->Evaluators->Arbitration->REVIEW_READY chain is
			// long done by the time a story reaches DONE; the only thing
			// left for this orchestrator to do is the Phase 8 retro stage —
			// a sibling stage to processStory below, not a continuation of
			// it (see processRetro's own doc comment for why it re-derives
			// the pipeline's tasks via ensureEvaluators/ensureArbitration
			// rather than assuming processStory has already run recently).
			o.processRetro(ctx, st)
			continue
		}
		o.processStory(ctx, st)
	}
}

func (o *StoryOrchestrator) logf(msg string, args ...any) {
	if o.Logger != nil {
		o.Logger.Warn(msg, args...)
	}
}

// processStory advances a single story by at most one stage per tick (it
// returns as soon as it finds a stage that isn't ready to progress yet — the
// next tick picks up where this one left off).
func (o *StoryOrchestrator) processStory(ctx context.Context, st *story.Story) {
	root, err := o.Store.GetTask(st.RootTaskID)
	if err != nil {
		o.logf("story orchestrator: get root task", "storyID", st.ID, "rootTaskID", st.RootTaskID, "error", err)
		return
	}
	root = o.autoAccept(st, root)
	if root.State != task.StateCompleted {
		return // Builder hasn't reached COMPLETED yet (still running, or not yet READY to auto-accept)
	}

	// Stage 1: Builder -> Evaluators (+ story -> VALIDATING).
	evaluators, ok := o.ensureEvaluators(ctx, st, root)
	if !ok {
		return // not all 4 could be found/created yet; retry next tick
	}

	// Auto-accept each evaluator that's reached READY, emit per-evaluator
	// verdicts, and check whether all 4 are done.
	allDone := true
	for i, ev := range evaluators {
		ev = o.autoAccept(st, ev)
		evaluators[i] = ev
		o.maybeEmitVerdict(st, ev)
		if ev.State != task.StateCompleted {
			allDone = false
		}
	}
	if !allDone {
		return
	}

	// Stage 2: Evaluators -> Arbitration.
	arbitration, ok := o.ensureArbitration(ctx, st, evaluators)
	if !ok {
		return
	}
	arbitration = o.autoAccept(st, arbitration)

	// Stage 3: Arbitration -> REVIEW_READY.
	if arbitration.State == task.StateCompleted {
		o.finalizeArbitration(st, arbitration)
	}
}

// autoAccept transitions t from READY to COMPLETED if that's its current
// state, using the same state-machine-respecting write internal/api's
// acceptTask uses (Store.UpdateTaskState wraps storage.DB.UpdateTaskStateBy,
// which validates task.ValidTransition and writes the state_change event
// atomically — not a raw/unchecked write). Returns t unchanged if it wasn't
// READY (including if it's already COMPLETED, or still RUNNING/BLOCKED/etc.)
// or if the update failed.
//
// This is the mechanism that makes the story-level accept-gate
// (POST /api/stories/{id}/accept) the *only* human/chatbot interaction
// required to drive a story from a completed Builder run all the way to
// REVIEW_READY: without it, a human would have to separately call
// POST /api/tasks/{id}/accept on the Builder task, each of the 4 Evaluator
// tasks, and the Arbitration task, since none of those top-level tasks can
// reach COMPLETED any other way (see this file's package doc comment).
// Callers only ever pass tasks they've already established are part of a
// specific story's pipeline — the root task a story tracks via
// st.RootTaskID, or a role-matched dependent discovered by
// ensureEvaluators/ensureArbitration — so this never touches an unrelated
// READY task sitting outside any story's pipeline.
func (o *StoryOrchestrator) autoAccept(st *story.Story, t *task.Task) *task.Task {
	if t.State != task.StateReady {
		return t
	}
	if err := o.Store.UpdateTaskState(t.ID, task.StateCompleted); err != nil {
		o.logf("story orchestrator: auto-accept", "storyID", st.ID, "taskID", t.ID, "error", err)
		return t
	}
	accepted := *t
	accepted.State = task.StateCompleted
	return &accepted
}

// ensureEvaluators returns the 4 Evaluator tasks fanned out from root,
// spawning any missing ones. Idempotency is structural, not a marker on the
// story: it looks at root's actual dependents and checks which of
// evaluatorRoles are already represented, so calling this repeatedly for the
// same story never spawns duplicates (test (b) in the phase description) —
// even across a process restart, unlike a purely in-memory guard would be.
// Returns ok=false if any missing evaluator couldn't be created this tick
// (transient store error); the caller retries on the next tick.
func (o *StoryOrchestrator) ensureEvaluators(ctx context.Context, st *story.Story, root *task.Task) ([]*task.Task, bool) {
	dependents, err := o.Store.ListDependents(root.ID)
	if err != nil {
		o.logf("story orchestrator: list root dependents", "storyID", st.ID, "error", err)
		return nil, false
	}
	found := make(map[string]*task.Task, len(evaluatorRoles))
	for _, d := range dependents {
		for _, r := range evaluatorRoles {
			if d.Agent.Role == r {
				found[r] = d
			}
		}
	}

	spawnedAny := false
	for _, r := range evaluatorRoles {
		if _, ok := found[r]; ok {
			continue
		}
		nt, err := o.spawnRoleTask(ctx, fmt.Sprintf("%s: %s", r, st.Name), r, []string{root.ID}, root,
			fmt.Sprintf("Evaluate the changes made by task %s against the %q dimension for story %q.\n\nStory spec:\n%s", root.ID, r, st.Name, st.Spec))
		if err != nil {
			o.logf("story orchestrator: spawn evaluator", "storyID", st.ID, "role", r, "error", err)
			continue
		}
		found[r] = nt
		spawnedAny = true
	}

	if spawnedAny {
		st.Status = "VALIDATING"
		if err := o.Store.UpdateStory(st); err != nil {
			o.logf("story orchestrator: update story to VALIDATING", "storyID", st.ID, "error", err)
		}
	}

	if len(found) != len(evaluatorRoles) {
		return nil, false
	}
	ordered := make([]*task.Task, len(evaluatorRoles))
	for i, r := range evaluatorRoles {
		ordered[i] = found[r]
	}
	return ordered, true
}

// ensureArbitration returns the single Arbitration task depending on all 4
// evaluators, spawning it if it doesn't exist yet. Idempotency is again
// structural: it looks for an existing "planner"-role dependent of the first
// evaluator task that depends on every evaluator ID, rather than relying on
// story.Status (which a human can freely rewrite via PUT /api/stories/{id}).
func (o *StoryOrchestrator) ensureArbitration(ctx context.Context, st *story.Story, evaluators []*task.Task) (*task.Task, bool) {
	ids := make([]string, len(evaluators))
	for i, ev := range evaluators {
		ids[i] = ev.ID
	}

	dependents, err := o.Store.ListDependents(evaluators[0].ID)
	if err != nil {
		o.logf("story orchestrator: list evaluator dependents", "storyID", st.ID, "error", err)
		return nil, false
	}
	for _, d := range dependents {
		if d.Agent.Role == arbitrationRole && dependsOnAll(d, ids) {
			return d, true
		}
	}

	instructions := fmt.Sprintf(
		"Arbitrate the 4 evaluator verdicts for story %q (task %s). Read each evaluator task's summary/events "+
			"and decide whether the story is ready to ship. Acceptance criteria:\n%s",
		st.Name, st.ID, formatAcceptanceCriteria(st.AcceptanceCriteria))
	nt, err := o.spawnRoleTask(ctx, "Arbitration: "+st.Name, arbitrationRole, ids, evaluators[0], instructions)
	if err != nil {
		o.logf("story orchestrator: spawn arbitration", "storyID", st.ID, "error", err)
		return nil, false
	}
	return nt, true
}

// finalizeArbitration handles the Arbitration task reaching COMPLETED: it
// emits KindArbitrationDecided and moves the story to REVIEW_READY or
// NEEDS_FIX, depending on whether the arbitration task recorded a structured
// KindVerdictReported event (AgentChannel.ReportVerdict). An arbitration
// task that never calls report_verdict defaults to REVIEW_READY, preserving
// the prior behavior for agents that don't report a structured verdict.
//
// Gated on st.Status == "VALIDATING" so repeated ticks (or a story a human
// already advanced past REVIEW_READY) don't re-emit the event or re-write the
// status — this is the one place in the orchestrator where the story's own
// status field, not a structural dependents check, is the idempotency guard,
// because by this stage there's nothing further to check structurally: the
// Arbitration task is the last task in the chain, so "does a subsequent task
// exist" isn't an available signal.
func (o *StoryOrchestrator) finalizeArbitration(st *story.Story, arbitration *task.Task) {
	if st.Status != "VALIDATING" {
		return
	}

	approved, hasVerdict := o.arbitrationVerdict(st, arbitration)

	payload, _ := json.Marshal(struct {
		TaskID   string `json:"task_id"`
		Summary  string `json:"summary"`
		Approved *bool  `json:"approved,omitempty"`
	}{TaskID: arbitration.ID, Summary: arbitration.Summary, Approved: optionalBool(approved, hasVerdict)})
	if err := o.Store.CreateEvent(&event.Event{
		TaskID:  st.ID,
		Kind:    event.KindArbitrationDecided,
		Actor:   event.ActorSystem,
		Payload: payload,
	}); err != nil {
		o.logf("story orchestrator: emit arbitration_decided", "storyID", st.ID, "error", err)
	}

	// Documented simplification (Phase 7b) closed: a structured
	// KindVerdictReported event (AgentChannel.ReportVerdict) on the
	// arbitration task, if present, now decides REVIEW_READY vs NEEDS_FIX.
	// An arbitration task that never calls report_verdict (hasVerdict ==
	// false) preserves the prior unconditional REVIEW_READY behavior — a
	// human or chatbot can still manually set NEEDS_FIX via the existing
	// PUT /api/stories/{id} for a story arbitrated by an agent that doesn't
	// report a structured verdict.
	if hasVerdict && !approved {
		st.Status = "NEEDS_FIX"
	} else {
		st.Status = "REVIEW_READY"
	}
	if err := o.Store.UpdateStory(st); err != nil {
		o.logf("story orchestrator: update story status after arbitration", "storyID", st.ID, "status", st.Status, "error", err)
	}
}

// arbitrationVerdict reads the arbitration task's own event stream for a
// KindVerdictReported event (AgentChannel.ReportVerdict) and returns its
// approved value. hasVerdict is false if no such event was ever recorded —
// callers must treat that as "no structured verdict was reported", not as
// an implicit rejection.
func (o *StoryOrchestrator) arbitrationVerdict(st *story.Story, arbitration *task.Task) (approved bool, hasVerdict bool) {
	events, err := o.Store.ListEvents(arbitration.ID, 0)
	if err != nil {
		o.logf("story orchestrator: list arbitration task events", "storyID", st.ID, "taskID", arbitration.ID, "error", err)
		return false, false
	}
	for _, e := range events {
		if e.Kind != event.KindVerdictReported {
			continue
		}
		var payload struct {
			Approved bool `json:"approved"`
		}
		if json.Unmarshal(e.Payload, &payload) == nil {
			approved = payload.Approved
			hasVerdict = true
		}
	}
	return approved, hasVerdict
}

// optionalBool returns a pointer to v if present is true, nil otherwise —
// used so KindArbitrationDecided's payload omits "approved" entirely
// (rather than encoding a misleading false) when no structured verdict was
// ever reported.
func optionalBool(v bool, present bool) *bool {
	if !present {
		return nil
	}
	return &v
}

// processRetro is the Phase 8 retro stage: it runs for every story the Tick
// loop has already observed at status DONE (see Tick above). Unlike
// processStory's stages, it never spawns or advances a Builder/Evaluator/
// Arbitration task itself — it only *looks up* the already-completed
// Evaluators/Arbitration (via the read-only findEvaluators/findArbitration
// below, not the spawning ensureEvaluators/ensureArbitration processStory
// uses) to find the Arbitration task the retro task should depend on. A real
// story never reaches DONE without that pipeline already having completed
// (DONE only follows REVIEW_READY, which only follows a completed
// Arbitration), so this is normally an immediate hit; if it's ever not
// found — e.g. this exact story/task shape was constructed directly at DONE
// without ever running the pipeline — processRetro simply has nothing to do
// yet and retries next tick, the same as any other transient "not ready"
// return in this file. This keeps the retro stage a strict sibling to the
// existing chain: it can never backfill or mutate Builder/Evaluator/
// Arbitration state, only react to it once it's already settled.
//
// Once the Arbitration task is found, this checks, structurally, whether a
// retro-role task already depends on it (mirroring ensureArbitration's own
// "does a role-matched dependent already exist" idempotency check) —
// spawning one only if not — and once that retro task reaches COMPLETED
// (auto-accepted from READY exactly like every other task in this pipeline),
// hands off to finalizeRetro to emit KindRetroCaptured.
func (o *StoryOrchestrator) processRetro(ctx context.Context, st *story.Story) {
	root, err := o.Store.GetTask(st.RootTaskID)
	if err != nil {
		o.logf("story orchestrator: retro: get root task", "storyID", st.ID, "rootTaskID", st.RootTaskID, "error", err)
		return
	}

	evaluators, ok := o.findEvaluators(root.ID)
	if !ok {
		return // pipeline not (yet) fully settled; nothing for retro to attach to
	}
	arbitration, ok := o.findArbitration(evaluators)
	if !ok {
		return
	}

	dependents, err := o.Store.ListDependents(arbitration.ID)
	if err != nil {
		o.logf("story orchestrator: retro: list arbitration dependents", "storyID", st.ID, "error", err)
		return
	}
	var retro *task.Task
	for _, d := range dependents {
		if d.Agent.Role == retroRole {
			retro = d
			break
		}
	}
	if retro == nil {
		instructions := o.buildRetroInstructions(st, root, arbitration)
		nt, err := o.spawnRoleTask(ctx, "Retro: "+st.Name, retroRole, []string{arbitration.ID}, root, instructions)
		if err != nil {
			o.logf("story orchestrator: retro: spawn", "storyID", st.ID, "error", err)
			return
		}
		retro = nt
	}

	retro = o.autoAccept(st, retro)
	if retro.State != task.StateCompleted {
		return // still running, or not yet READY to auto-accept
	}
	o.finalizeRetro(st, retro)
}

// findEvaluators returns root's 4 evaluator-role dependents if all of them
// already exist — read-only, unlike ensureEvaluators, which spawns any
// missing ones. processRetro deliberately never spawns Builder-pipeline
// tasks itself (see that function's doc comment); if the 4 evaluators
// aren't all found, ok is false and processRetro has nothing to do this
// tick.
func (o *StoryOrchestrator) findEvaluators(rootID string) ([]*task.Task, bool) {
	dependents, err := o.Store.ListDependents(rootID)
	if err != nil {
		return nil, false
	}
	found := make(map[string]*task.Task, len(evaluatorRoles))
	for _, d := range dependents {
		for _, r := range evaluatorRoles {
			if d.Agent.Role == r {
				found[r] = d
			}
		}
	}
	if len(found) != len(evaluatorRoles) {
		return nil, false
	}
	ordered := make([]*task.Task, len(evaluatorRoles))
	for i, r := range evaluatorRoles {
		ordered[i] = found[r]
	}
	return ordered, true
}

// findArbitration returns the "planner"-role task depending on every
// evaluator in evaluators, if one exists — the read-only counterpart to
// ensureArbitration, for the same reason findEvaluators is read-only (see
// processRetro's doc comment).
func (o *StoryOrchestrator) findArbitration(evaluators []*task.Task) (*task.Task, bool) {
	ids := make([]string, len(evaluators))
	for i, ev := range evaluators {
		ids[i] = ev.ID
	}
	dependents, err := o.Store.ListDependents(evaluators[0].ID)
	if err != nil {
		return nil, false
	}
	for _, d := range dependents {
		if d.Agent.Role == arbitrationRole && dependsOnAll(d, ids) {
			return d, true
		}
	}
	return nil, false
}

// buildRetroInstructions assembles the retro task's Instructions from the
// story's full task tree (subtasks + DAG dependents, the same walk
// GET /api/stories/{id}/task-tree does in internal/api/stories.go), each
// task's accumulated cost and highest escalation rung, the currently active
// role_configs for every role encountered, and the story's own event stream
// (evaluator verdicts, the arbitration decision) — everything a retro needs
// to reflect on what happened and judge whether any role's config is worth
// revising. Kept to one story at a time, no cross-story/epic aggregation
// (that's explicitly out of scope for this phase).
func (o *StoryOrchestrator) buildRetroInstructions(st *story.Story, root, arbitration *task.Task) string {
	var b strings.Builder
	fmt.Fprintf(&b, "You are running a retrospective for story %q (id %s), which just reached DONE.\n\n", st.Name, st.ID)
	fmt.Fprintf(&b, "Story spec:\n%s\n\n", st.Spec)
	fmt.Fprintf(&b, "Acceptance criteria:\n%s\n", formatAcceptanceCriteria(st.AcceptanceCriteria))

	nodes := o.taskTree(root.ID)
	b.WriteString("\nTask tree (name, role, state, provider/model, accumulated cost, highest escalation rung):\n")
	totalCost := 0.0
	rolesSeen := map[string]bool{}
	for _, t := range nodes {
		execs, err := o.Store.ListExecutions(t.ID)
		if err != nil {
			o.logf("story orchestrator: retro: list executions", "storyID", st.ID, "taskID", t.ID, "error", err)
		}
		cost := 0.0
		maxRung := 0
		for _, e := range execs {
			cost += e.CostUSD
			if e.EscalationRung > maxRung {
				maxRung = e.EscalationRung
			}
		}
		totalCost += cost
		fmt.Fprintf(&b, "- %s (role=%q, state=%s, agent=%s/%s, cost=$%.4f, max_escalation_rung=%d)\n",
			t.Name, t.Agent.Role, t.State, t.Agent.Type, t.Agent.Model, cost, maxRung)
		if t.Agent.Role != "" {
			rolesSeen[t.Agent.Role] = true
		}
	}
	fmt.Fprintf(&b, "\nTotal estimated cost across the story's task tree: $%.4f\n", totalCost)

	b.WriteString("\nActive role configs for roles involved in this story:\n")
	for r := range rolesSeen {
		row, err := o.Store.GetActiveRoleConfig(r)
		if err != nil {
			fmt.Fprintf(&b, "- %s: no active role config\n", r)
			continue
		}
		fmt.Fprintf(&b, "- %s (active version %d):\n%s\n", r, row.Version, row.ConfigJSON)
	}

	if events, err := o.Store.ListEvents(st.ID, 0); err != nil {
		o.logf("story orchestrator: retro: list story events", "storyID", st.ID, "error", err)
	} else if len(events) > 0 {
		b.WriteString("\nStory events (evaluator verdicts, arbitration decision, human accept):\n")
		for _, e := range events {
			fmt.Fprintf(&b, "- [%s] %s\n", e.Kind, string(e.Payload))
		}
	}

	b.WriteString("\nYour job: reflect on what happened across this story -- the roles/configs involved, " +
		"escalations, cost, and any evaluator/arbitration feedback -- and propose concrete improvements. For each " +
		"role whose config you judge worth revising, call propose_role_config with a complete, improved role " +
		"config (role name, and whichever of system_prompt/escalation_ladder/tools/sandbox_kind/" +
		"default_budget_usd you're changing). Only propose changes you can justify from what actually happened " +
		"in this story; it is fine to propose zero changes if nothing stands out. Before finishing, call " +
		"report_summary with the qualitative lessons learned (what worked, what didn't, and why) -- that is this " +
		"retro's most important output even when you propose no config changes.")

	return b.String()
}

// taskTree walks the task graph realizing a story starting at rootID,
// following both parent_task_id children (ListSubtasks) and depends_on edges
// in either direction (ListDependents) — the identical BFS
// GET /api/stories/{id}/task-tree performs in internal/api/stories.go,
// reimplemented here against StoryStore so the retro stage can assemble the
// same view without an HTTP round-trip to its own server (per this phase's
// guidance to use Store methods directly). Errors from either listing call
// are logged and treated as "no further edges from this node" rather than
// aborting the whole walk, matching the API handler's tolerance for a
// dangling reference.
func (o *StoryOrchestrator) taskTree(rootID string) []*task.Task {
	visited := map[string]bool{}
	queue := []string{rootID}
	var out []*task.Task
	for len(queue) > 0 {
		id := queue[0]
		queue = queue[1:]
		if visited[id] {
			continue
		}
		visited[id] = true

		t, err := o.Store.GetTask(id)
		if err != nil {
			continue // referenced task no longer resolves; skip it, don't abort the walk
		}
		out = append(out, t)

		if children, err := o.Store.ListSubtasks(id); err == nil {
			for _, c := range children {
				if !visited[c.ID] {
					queue = append(queue, c.ID)
				}
			}
		}
		if deps, err := o.Store.ListDependents(id); err == nil {
			for _, d := range deps {
				if !visited[d.ID] {
					queue = append(queue, d.ID)
				}
			}
		}
	}
	return out
}

// finalizeRetro emits event.KindRetroCaptured, attached to the story's ID
// (matching this orchestrator's convention for planning-layer ceremony
// events — KindEvalVerdict/KindArbitrationDecided are attached the same
// way), once the retro task reaches COMPLETED. The payload aggregates every
// event.KindRoleConfigProposed the retro task itself recorded (one per
// propose_role_config call — see storeChannel.ProposeRoleConfig in
// internal/executor/channel.go) into a single list of {role, version}, plus
// the retro task's own reported summary (the "capturing lessons" half of
// this phase's brief) — this is the final mechanism in the harness's
// self-improvement loop: retro produces drafts here, a human activates them
// via the unchanged POST /api/roles/{role}/activate, and future dispatches
// use the improved config.
//
// Guarded by handledRetro (in-memory, per-process — see that field's doc
// comment) so a story sitting at DONE forever doesn't re-emit this event on
// every subsequent tick.
func (o *StoryOrchestrator) finalizeRetro(st *story.Story, retro *task.Task) {
	if o.handledRetro == nil {
		o.handledRetro = make(map[string]bool)
	}
	if o.handledRetro[retro.ID] {
		return
	}
	o.handledRetro[retro.ID] = true

	type proposedRoleConfig struct {
		Role    string `json:"role"`
		Version int    `json:"version"`
	}
	var proposals []proposedRoleConfig
	events, err := o.Store.ListEvents(retro.ID, 0)
	if err != nil {
		o.logf("story orchestrator: retro: list retro task events", "storyID", st.ID, "taskID", retro.ID, "error", err)
	}
	for _, e := range events {
		if e.Kind != event.KindRoleConfigProposed {
			continue
		}
		var p proposedRoleConfig
		if json.Unmarshal(e.Payload, &p) == nil {
			proposals = append(proposals, p)
		}
	}

	payload, _ := json.Marshal(struct {
		TaskID    string               `json:"task_id"`
		Proposals []proposedRoleConfig `json:"proposals"`
		Summary   string               `json:"summary"`
	}{TaskID: retro.ID, Proposals: proposals, Summary: retro.Summary})
	if err := o.Store.CreateEvent(&event.Event{
		TaskID:  st.ID,
		Kind:    event.KindRetroCaptured,
		Actor:   event.ActorSystem,
		Payload: payload,
	}); err != nil {
		o.logf("story orchestrator: retro: emit retro_captured", "storyID", st.ID, "error", err)
	}
}

// maybeEmitVerdict records a KindEvalVerdict event, attached to the story's
// ID (not the evaluator task's ID), the first time a given evaluator task is
// observed COMPLETED. Attaching to the story ID — the same choice
// finalizeArbitration makes for KindArbitrationDecided — means a single
// GET /api/stories/{id}/events call surfaces every verdict for a story,
// rather than requiring a client to separately fetch each evaluator task's
// own event stream and reassemble them; events.task_id has no enforced FK
// (see internal/event's doc comment), so this is exactly the tolerance the
// 7a phase already built in anticipation of this use.
func (o *StoryOrchestrator) maybeEmitVerdict(st *story.Story, ev *task.Task) {
	if ev.State != task.StateCompleted {
		return
	}
	if o.handledVerdicts == nil {
		o.handledVerdicts = make(map[string]bool)
	}
	if o.handledVerdicts[ev.ID] {
		return
	}
	o.handledVerdicts[ev.ID] = true

	payload, _ := json.Marshal(struct {
		TaskID  string `json:"task_id"`
		Role    string `json:"role"`
		Summary string `json:"summary"`
	}{TaskID: ev.ID, Role: ev.Agent.Role, Summary: ev.Summary})
	if err := o.Store.CreateEvent(&event.Event{
		TaskID:  st.ID,
		Kind:    event.KindEvalVerdict,
		Actor:   event.ActorSystem,
		Payload: payload,
	}); err != nil {
		o.logf("story orchestrator: emit eval_verdict", "storyID", st.ID, "taskID", ev.ID, "error", err)
	}
}

// spawnRoleTask creates a new role-typed, top-level task (no ParentTaskID —
// these are DAG siblings via DependsOn, not delegated subtasks; see
// internal/executor.Pool.cascadeFail's doc comment for why that distinction
// matters) and submits it to the pool. Agent.Type/Model are left empty so
// Phase 5's role-based dispatch resolves them from the role's escalation
// ladder on first dispatch (internal/executor.Pool.execute()); if no active
// role_configs row exists for the role, that same code path logs a warning
// and dispatches without role resolution — an accepted degraded mode we
// don't special-case here either.
func (o *StoryOrchestrator) spawnRoleTask(ctx context.Context, name, roleName string, dependsOn []string, template *task.Task, instructions string) (*task.Task, error) {
	nt := &task.Task{
		ID:            uuid.NewString(),
		Name:          name,
		Project:       template.Project,
		RepositoryURL: template.RepositoryURL,
		Agent: task.AgentConfig{
			Role:         roleName,
			Instructions: instructions,
		},
		Priority:  task.PriorityNormal,
		Tags:      []string{"story-orchestrator"},
		DependsOn: dependsOn,
		Retry:     task.RetryConfig{MaxAttempts: 1, Backoff: "exponential"},
		State:     task.StatePending,
	}
	if err := o.Store.CreateTask(nt); err != nil {
		return nil, err
	}
	if err := o.Store.UpdateTaskState(nt.ID, task.StateQueued); err != nil {
		return nil, err
	}
	nt.State = task.StateQueued
	if err := o.Pool.Submit(ctx, nt); err != nil {
		return nil, err
	}
	return nt, nil
}

// dependsOnAll reports whether t.DependsOn contains every ID in ids.
func dependsOnAll(t *task.Task, ids []string) bool {
	have := make(map[string]bool, len(t.DependsOn))
	for _, d := range t.DependsOn {
		have[d] = true
	}
	for _, id := range ids {
		if !have[id] {
			return false
		}
	}
	return true
}

// formatAcceptanceCriteria renders a story's acceptance criteria as a
// markdown bullet list, or a placeholder line if there are none.
func formatAcceptanceCriteria(criteria []string) string {
	if len(criteria) == 0 {
		return "(none specified)"
	}
	out := ""
	for _, c := range criteria {
		out += "- " + c + "\n"
	}
	return out
}