| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "testing" |
| 12 | |
| 13 | "reasonix/internal/event" |
| 14 | "reasonix/internal/provider" |
| 15 | "reasonix/internal/tool" |
| 16 | ) |
| 17 | |
| 18 | func planFor(t *testing.T, items ...fleetTaskItem) (fleetPlan, error) { |
| 19 | t.Helper() |
| 20 | return newFleetPlan(items, false) |
| 21 | } |
| 22 | |
| 23 | func TestFleetPlanRejectsBrokenGraphsBeforeAnythingRuns(t *testing.T) { |
| 24 | for name, tc := range map[string]struct { |
| 25 | items []fleetTaskItem |
| 26 | want string |
| 27 | }{ |
| 28 | "duplicate id": { |
| 29 | items: []fleetTaskItem{{ID: "a"}, {ID: "a"}}, |
| 30 | want: "already used", |
| 31 | }, |
| 32 | "unknown dependency": { |
| 33 | items: []fleetTaskItem{{ID: "a"}, {ID: "b", DependsOn: []string{"nope"}}}, |
| 34 | want: "matches no task id", |
| 35 | }, |
| 36 | "self dependency": { |
| 37 | items: []fleetTaskItem{{ID: "a", DependsOn: []string{"a"}}, {ID: "b"}}, |
| 38 | want: "depends_on itself", |
| 39 | }, |
| 40 | "cycle": { |
| 41 | items: []fleetTaskItem{ |
| 42 | {ID: "a", DependsOn: []string{"c"}}, |
| 43 | {ID: "b", DependsOn: []string{"a"}}, |
| 44 | {ID: "c", DependsOn: []string{"b"}}, |
| 45 | }, |
| 46 | want: "cycle", |
| 47 | }, |
| 48 | } { |
| 49 | if _, err := planFor(t, tc.items...); err == nil || !strings.Contains(err.Error(), tc.want) { |
| 50 | t.Errorf("%s: err = %v, want one mentioning %q", name, err, tc.want) |
| 51 | } |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | func TestFleetPlanOrdersTransitiveDependents(t *testing.T) { |
| 56 | plan, err := planFor(t, |
| 57 | fleetTaskItem{ID: "research"}, |
| 58 | fleetTaskItem{ID: "backend", DependsOn: []string{"research"}}, |
| 59 | fleetTaskItem{ID: "frontend", DependsOn: []string{"research"}}, |
| 60 | fleetTaskItem{ID: "integration", DependsOn: []string{"backend", "frontend"}}, |
| 61 | ) |
| 62 | if err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | if !plan.ordered(0, 3) { |
| 66 | t.Error("integration transitively depends on research and must be ordered against it") |
| 67 | } |
| 68 | if plan.ordered(1, 2) { |
| 69 | t.Error("backend and frontend share a dependency but not an order: they run in parallel") |
| 70 | } |
| 71 | if roots := plan.roots(); len(roots) != 1 || roots[0] != 0 { |
| 72 | t.Errorf("roots = %v, want just research", roots) |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | // The unlock: implement → review legitimately touch the same files, which a |
| 77 | // flat fleet could never express. |
| 78 | func TestFleetOrderedWritersMayShareWritePaths(t *testing.T) { |
| 79 | root := t.TempDir() |
| 80 | claim, err := NormalizeWritePaths(root, []string{"api"}) |
| 81 | if err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | ordered, err := planFor(t, |
| 85 | fleetTaskItem{ID: "implement"}, |
| 86 | fleetTaskItem{ID: "review", DependsOn: []string{"implement"}}, |
| 87 | ) |
| 88 | if err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | if err := ordered.validateConcurrentWriteClaims([]WritePathSet{claim, claim}); err != nil { |
| 92 | t.Fatalf("ordered writers must be allowed to share paths: %v", err) |
| 93 | } |
| 94 | |
| 95 | concurrent, err := planFor(t, fleetTaskItem{ID: "a"}, fleetTaskItem{ID: "b"}) |
| 96 | if err != nil { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | if err := concurrent.validateConcurrentWriteClaims([]WritePathSet{claim, claim}); err == nil { |
| 100 | t.Fatal("two writers that can run at once must still fail preflight on overlap") |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | func TestFleetConcurrentDirectoryClaimsPassPreflight(t *testing.T) { |
| 105 | root := t.TempDir() |
| 106 | if err := os.MkdirAll(filepath.Join(root, "src"), 0o755); err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | claim, err := NormalizeWritePaths(root, []string{"src/"}) |
| 110 | if err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | concurrent, err := planFor(t, fleetTaskItem{ID: "a"}, fleetTaskItem{ID: "b"}) |
| 114 | if err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | if err := concurrent.validateConcurrentWriteClaims([]WritePathSet{claim, claim}); err != nil { |
| 118 | t.Fatalf("concurrent directory claims must pass preflight: %v", err) |
| 119 | } |
| 120 | whole, err := WholeWorkspaceWriteClaim(root) |
| 121 | if err != nil { |
| 122 | t.Fatal(err) |
| 123 | } |
| 124 | if err := concurrent.validateConcurrentWriteClaims([]WritePathSet{whole, whole}); err != nil { |
| 125 | t.Fatalf("omitted write_paths must queue in the scheduler, not fail preflight: %v", err) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | func TestFleetPlanSkipsWholeDownstreamBranch(t *testing.T) { |
| 130 | plan, err := planFor(t, |
| 131 | fleetTaskItem{ID: "research"}, |
| 132 | fleetTaskItem{ID: "implement", DependsOn: []string{"research"}}, |
| 133 | fleetTaskItem{ID: "review", DependsOn: []string{"implement"}}, |
| 134 | fleetTaskItem{ID: "unrelated"}, |
| 135 | ) |
| 136 | if err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | results := make([]fleetItemResult, 4) |
| 140 | for i := range results { |
| 141 | results[i] = fleetItemResult{index: i, status: fleetItemPending} |
| 142 | } |
| 143 | results[0].status = fleetItemFailed |
| 144 | plan.skipDependents(results, 0) |
| 145 | |
| 146 | if results[1].status != fleetItemSkipped || results[2].status != fleetItemSkipped { |
| 147 | t.Fatalf("statuses = %q/%q, want the whole downstream branch skipped", results[1].status, results[2].status) |
| 148 | } |
| 149 | if !strings.Contains(results[2].err.Error(), `depends on "research"`) { |
| 150 | t.Fatalf("skip reason = %v, want it to name the broken dependency", results[2].err) |
| 151 | } |
| 152 | if results[3].status != fleetItemPending { |
| 153 | t.Error("an unrelated branch must not be skipped by another branch's failure") |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | // End to end: a dependent never runs when its dependency failed. |
| 158 | func TestFleetSkipsDependentsOfFailedTask(t *testing.T) { |
| 159 | root := t.TempDir() |
| 160 | prov := &fleetScriptedFailureProvider{} |
| 161 | reg := tool.NewRegistry() |
| 162 | reg.Add(fakeReadFileTool{}) |
| 163 | task := NewTaskTool(prov, nil, reg, 20, 0, 0, 0, 0, 0, 0, 0.0, "", "sys", nil, 0, "", "", nil). |
| 164 | WithTranscripts(mustSubagentStore(t), root, "base", "high"). |
| 165 | WithScheduler(NewSubagentScheduler(4, 4)) |
| 166 | fleet := NewFleetTool(task) |
| 167 | ctx := withCallContext(context.Background(), "fleet-call", event.Discard, nil, false) |
| 168 | |
| 169 | out, _ := fleet.Execute(ctx, json.RawMessage(`{"tasks":[ |
| 170 | {"id":"research","prompt":"FAIL research","read_only":true}, |
| 171 | {"id":"implement","prompt":"implement","depends_on":["research"],"write_paths":["api"]}, |
| 172 | {"id":"sibling","prompt":"sibling","read_only":true} |
| 173 | ]}`)) |
| 174 | |
| 175 | if !strings.Contains(out, "skipped") || !strings.Contains(out, `depends on "research"`) { |
| 176 | t.Fatalf("aggregate must report the dependent as skipped with its reason:\n%s", out) |
| 177 | } |
| 178 | if prov.ran("implement") { |
| 179 | t.Fatal("a dependent of a failed task must never start") |
| 180 | } |
| 181 | if !prov.ran("sibling") { |
| 182 | t.Fatal("an independent branch must still run when another branch fails") |
| 183 | } |
| 184 | } |
| 185 | |
| 186 | type fleetScriptedFailureProvider struct { |
| 187 | mu sync.Mutex |
| 188 | seen map[string]bool |
| 189 | } |
| 190 | |
| 191 | func (p *fleetScriptedFailureProvider) Name() string { return "fleet-failure" } |
| 192 | |
| 193 | func (p *fleetScriptedFailureProvider) ran(name string) bool { |
| 194 | p.mu.Lock() |
| 195 | defer p.mu.Unlock() |
| 196 | return p.seen[name] |
| 197 | } |
| 198 | |
| 199 | func (p *fleetScriptedFailureProvider) Stream(_ context.Context, req provider.Request) (<-chan provider.Chunk, error) { |
| 200 | last := "" |
| 201 | for _, m := range req.Messages { |
| 202 | if m.Role == provider.RoleUser { |
| 203 | last = m.Content |
| 204 | } |
| 205 | } |
| 206 | p.mu.Lock() |
| 207 | if p.seen == nil { |
| 208 | p.seen = map[string]bool{} |
| 209 | } |
| 210 | for _, name := range []string{"implement", "sibling"} { |
| 211 | if strings.Contains(last, name) { |
| 212 | p.seen[name] = true |
| 213 | } |
| 214 | } |
| 215 | p.mu.Unlock() |
| 216 | if strings.Contains(last, "FAIL") { |
| 217 | return nil, errors.New("scripted provider failure") |
| 218 | } |
| 219 | ch := make(chan provider.Chunk, 2) |
| 220 | ch <- provider.Chunk{Type: provider.ChunkText, Text: "done"} |
| 221 | ch <- provider.Chunk{Type: provider.ChunkDone} |
| 222 | close(ch) |
| 223 | return ch, nil |
| 224 | } |
| 225 |