summaryrefslogtreecommitdiff
path: root/internal/scheduler
diff options
context:
space:
mode:
Diffstat (limited to 'internal/scheduler')
-rw-r--r--internal/scheduler/story_orchestrator.go343
-rw-r--r--internal/scheduler/story_orchestrator_test.go254
2 files changed, 596 insertions, 1 deletions
diff --git a/internal/scheduler/story_orchestrator.go b/internal/scheduler/story_orchestrator.go
index 6ad0f86..b8e1b4c 100644
--- a/internal/scheduler/story_orchestrator.go
+++ b/internal/scheduler/story_orchestrator.go
@@ -47,6 +47,7 @@ import (
"encoding/json"
"fmt"
"log/slog"
+ "strings"
"time"
"github.com/google/uuid"
@@ -75,6 +76,17 @@ var evaluatorRoles = []string{
// its summary for an approve/reject verdict.
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.
@@ -86,6 +98,19 @@ type StoryStore interface {
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
@@ -119,6 +144,16 @@ type StoryOrchestrator struct {
// 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.
@@ -154,9 +189,20 @@ func (o *StoryOrchestrator) Tick(ctx context.Context) {
if st.RootTaskID == "" {
continue // no execution tree yet — nothing for this orchestrator to do
}
- if st.Status == "DONE" || st.Status == "CANCELLED" {
+ 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)
}
}
@@ -380,6 +426,301 @@ func (o *StoryOrchestrator) finalizeArbitration(st *story.Story, arbitration *ta
}
}
+// 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
diff --git a/internal/scheduler/story_orchestrator_test.go b/internal/scheduler/story_orchestrator_test.go
index bee326d..b811206 100644
--- a/internal/scheduler/story_orchestrator_test.go
+++ b/internal/scheduler/story_orchestrator_test.go
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
+ "strings"
"sync"
"testing"
@@ -23,6 +24,12 @@ type fakeStoryStore struct {
stories map[string]*story.Story
tasks map[string]*task.Task
events []*event.Event
+
+ // executions and activeRoleConfigs back the Phase 8 retro stage's
+ // context-gathering (ListExecutions/GetActiveRoleConfig) — keyed the
+ // same way the real storage.DB is (by task ID / role name).
+ executions map[string][]*storage.Execution
+ activeRoleConfigs map[string]*storage.RoleConfigRow
}
func newFakeStoryStore() *fakeStoryStore {
@@ -107,10 +114,52 @@ func (f *fakeStoryStore) CreateEvent(e *event.Event) error {
f.mu.Lock()
defer f.mu.Unlock()
e.ID = uuid.NewString()
+ e.Seq = int64(len(f.events)) + 1
f.events = append(f.events, e)
return nil
}
+func (f *fakeStoryStore) ListSubtasks(parentID string) ([]*task.Task, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ var out []*task.Task
+ for _, t := range f.tasks {
+ if t.ParentTaskID == parentID {
+ cp := *t
+ out = append(out, &cp)
+ }
+ }
+ return out, nil
+}
+
+func (f *fakeStoryStore) ListExecutions(taskID string) ([]*storage.Execution, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.executions[taskID], nil
+}
+
+func (f *fakeStoryStore) GetActiveRoleConfig(roleName string) (*storage.RoleConfigRow, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ row, ok := f.activeRoleConfigs[roleName]
+ if !ok {
+ return nil, fmt.Errorf("no active role config for %q", roleName)
+ }
+ return row, nil
+}
+
+func (f *fakeStoryStore) ListEvents(taskID string, sinceSeq int64) ([]*event.Event, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ var out []*event.Event
+ for _, e := range f.events {
+ if e.TaskID == taskID && e.Seq > sinceSeq {
+ out = append(out, e)
+ }
+ }
+ return out, nil
+}
+
func (f *fakeStoryStore) setTaskState(id string, s task.State) {
f.mu.Lock()
defer f.mu.Unlock()
@@ -767,3 +816,208 @@ func TestStoryOrchestrator_EndToEnd(t *testing.T) {
t.Fatalf("expected exactly 1 arbitration_decided event, got %d", len(decided))
}
}
+
+// seedDoneStory wires up a story whose entire Builder->Evaluators->
+// Arbitration pipeline has already completed (mirroring what would actually
+// be true of any real story by the time it reaches DONE) and whose status
+// is DONE, returning the fakeStoryStore, story, root, evaluators, and
+// arbitration task. Used by the Phase 8 retro-stage tests below.
+func seedDoneStory(t *testing.T) (*fakeStoryStore, *story.Story, *task.Task, []*task.Task, *task.Task) {
+ t.Helper()
+ store, st, evaluators := seedStoryWithEvaluators(t, task.StateCompleted)
+ st.Status = "DONE"
+
+ ids := make([]string, len(evaluators))
+ for i, ev := range evaluators {
+ ids[i] = ev.ID
+ }
+ arb := &task.Task{
+ ID: "arbitration-1",
+ Name: "Arbitration",
+ Agent: task.AgentConfig{Role: arbitrationRole},
+ DependsOn: ids,
+ State: task.StateCompleted,
+ Summary: "ship it",
+ }
+ store.tasks[arb.ID] = arb
+
+ root, err := store.GetTask(st.RootTaskID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return store, st, root, evaluators, arb
+}
+
+// TestStoryOrchestrator_Retro_SpawnsRetroTask_WhenStoryDone is verification
+// item (a): a story reaching DONE spawns exactly one retro-role task
+// depending on the arbitration task, whose instructions contain the story's
+// context (name and ID, at minimum).
+func TestStoryOrchestrator_Retro_SpawnsRetroTask_WhenStoryDone(t *testing.T) {
+ store, st, _, _, arb := seedDoneStory(t)
+
+ pool := &fakePool{}
+ orch := &StoryOrchestrator{Store: store, Pool: pool}
+ orch.Tick(context.Background())
+
+ retros := store.dependentsWithRole(arb.ID, retroRole)
+ if len(retros) != 1 {
+ t.Fatalf("expected exactly 1 retro task depending on the arbitration task, got %d", len(retros))
+ }
+ retro := retros[0]
+ if retro.ParentTaskID != "" {
+ t.Errorf("retro task should be a DAG sibling (no ParentTaskID), got %q", retro.ParentTaskID)
+ }
+ if retro.State != task.StateQueued {
+ t.Errorf("retro task state: want QUEUED, got %v", retro.State)
+ }
+ if !strings.Contains(retro.Agent.Instructions, st.Name) || !strings.Contains(retro.Agent.Instructions, st.ID) {
+ t.Errorf("retro instructions should contain the story's name/ID, got: %s", retro.Agent.Instructions)
+ }
+}
+
+// TestStoryOrchestrator_Retro_DoesNotDuplicate is verification item (b):
+// checking the same DONE story again does not spawn a second retro task.
+func TestStoryOrchestrator_Retro_DoesNotDuplicate(t *testing.T) {
+ store, _, _, _, arb := seedDoneStory(t)
+
+ pool := &fakePool{}
+ orch := &StoryOrchestrator{Store: store, Pool: pool}
+ orch.Tick(context.Background())
+ orch.Tick(context.Background())
+ orch.Tick(context.Background())
+
+ retros := store.dependentsWithRole(arb.ID, retroRole)
+ if len(retros) != 1 {
+ t.Fatalf("expected still exactly 1 retro task after repeated ticks, got %d", len(retros))
+ }
+}
+
+// TestStoryOrchestrator_Retro_DoesNothing_UntilPipelineSettled proves
+// processRetro never backfills a Builder/Evaluator/Arbitration task itself:
+// a story marked DONE whose evaluators/arbitration don't actually exist yet
+// (a hypothetical/malformed case — a real story never reaches DONE any
+// other way) leaves the task tree untouched, exactly like
+// TestStoryOrchestrator_SkipsTerminalStories already proves for the
+// evaluator-spawning side of this.
+func TestStoryOrchestrator_Retro_DoesNothing_UntilPipelineSettled(t *testing.T) {
+ store := newFakeStoryStore()
+ root := builderTask("builder-1", task.StateCompleted)
+ store.tasks[root.ID] = root
+ st := newStoryWithRoot("story-1", root.ID, "DONE")
+ store.stories[st.ID] = st
+
+ pool := &fakePool{}
+ orch := &StoryOrchestrator{Store: store, Pool: pool}
+ orch.Tick(context.Background())
+
+ deps, _ := store.ListDependents(root.ID)
+ if len(deps) != 0 {
+ t.Fatalf("expected no tasks spawned when the pipeline never actually ran, got %d", len(deps))
+ }
+}
+
+// TestStoryOrchestrator_Retro_EmitsRetroCaptured_WhenRetroTaskCompletes is
+// verification item (d): once the retro task reaches COMPLETED (via
+// auto-accept from READY, exactly like every other task in this pipeline),
+// KindRetroCaptured is emitted attached to the story's ID, aggregating every
+// role_config_proposed event the retro task itself recorded plus its
+// reported summary.
+func TestStoryOrchestrator_Retro_EmitsRetroCaptured_WhenRetroTaskCompletes(t *testing.T) {
+ store, st, _, _, arb := seedDoneStory(t)
+
+ pool := &fakePool{}
+ orch := &StoryOrchestrator{Store: store, Pool: pool}
+ // Tick 1: spawns the retro task.
+ orch.Tick(context.Background())
+
+ retros := store.dependentsWithRole(arb.ID, retroRole)
+ if len(retros) != 1 {
+ t.Fatalf("expected 1 retro task, got %d", len(retros))
+ }
+ retro := retros[0]
+
+ // Simulate the retro agent having called propose_role_config twice
+ // (recording KindRoleConfigProposed events attached to its own task ID,
+ // exactly as storeChannel.ProposeRoleConfig does — see
+ // internal/executor/channel.go) before finishing with a summary.
+ mustMarshal := func(v any) json.RawMessage {
+ b, err := json.Marshal(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return b
+ }
+ store.CreateEvent(&event.Event{
+ TaskID: retro.ID,
+ Kind: event.KindRoleConfigProposed,
+ Actor: event.ActorAgent,
+ Payload: mustMarshal(struct {
+ Role string `json:"role"`
+ Version int `json:"version"`
+ }{Role: "builder", Version: 2}),
+ })
+ store.CreateEvent(&event.Event{
+ TaskID: retro.ID,
+ Kind: event.KindRoleConfigProposed,
+ Actor: event.ActorAgent,
+ Payload: mustMarshal(struct {
+ Role string `json:"role"`
+ Version int `json:"version"`
+ }{Role: "evaluator_quality", Version: 1}),
+ })
+ store.setTaskState(retro.ID, task.StateReady)
+ store.setTaskSummary(retro.ID, "The builder over-escalated twice; tightening its system prompt should help.")
+
+ // Tick 2: retro task auto-accepted to COMPLETED, retro_captured emitted.
+ orch.Tick(context.Background())
+
+ retroAfter, err := store.GetTask(retro.ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if retroAfter.State != task.StateCompleted {
+ t.Fatalf("retro task should be auto-accepted to COMPLETED, got %v", retroAfter.State)
+ }
+
+ captured := store.eventsOfKind(event.KindRetroCaptured)
+ if len(captured) != 1 {
+ t.Fatalf("expected exactly 1 retro_captured event, got %d", len(captured))
+ }
+ if captured[0].TaskID != st.ID {
+ t.Errorf("retro_captured should be attached to the story ID %q, got %q", st.ID, captured[0].TaskID)
+ }
+ var payload struct {
+ TaskID string `json:"task_id"`
+ Proposals []struct {
+ Role string `json:"role"`
+ Version int `json:"version"`
+ } `json:"proposals"`
+ Summary string `json:"summary"`
+ }
+ if err := json.Unmarshal(captured[0].Payload, &payload); err != nil {
+ t.Fatalf("unmarshal payload: %v", err)
+ }
+ if payload.TaskID != retro.ID {
+ t.Errorf("payload.task_id: want %q, got %q", retro.ID, payload.TaskID)
+ }
+ if len(payload.Proposals) != 2 {
+ t.Fatalf("expected 2 proposals in payload, got %d: %+v", len(payload.Proposals), payload.Proposals)
+ }
+ if payload.Proposals[0].Role != "builder" || payload.Proposals[0].Version != 2 {
+ t.Errorf("proposal[0] mismatch: %+v", payload.Proposals[0])
+ }
+ if payload.Proposals[1].Role != "evaluator_quality" || payload.Proposals[1].Version != 1 {
+ t.Errorf("proposal[1] mismatch: %+v", payload.Proposals[1])
+ }
+ if !strings.Contains(payload.Summary, "over-escalated") {
+ t.Errorf("expected the retro's summary in the payload, got %q", payload.Summary)
+ }
+
+ // Tick 3+: must not re-emit.
+ orch.Tick(context.Background())
+ orch.Tick(context.Background())
+ captured = store.eventsOfKind(event.KindRetroCaptured)
+ if len(captured) != 1 {
+ t.Fatalf("expected still exactly 1 retro_captured event after repeated ticks, got %d", len(captured))
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-16 22:22:54 +0000