| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "strconv" |
| 7 | "strings" |
| 8 | ) |
| 9 | |
| 10 | // fleetPlan is the validated dependency graph for one fleet call. Dependencies |
| 11 | // live here, on the graph, and never on a task spec: what a task is does not |
| 12 | // depend on what ran before it. Keeping them apart is what stops fleet from |
| 13 | // growing into a workflow language. |
| 14 | type fleetPlan struct { |
| 15 | ids []string |
| 16 | deps [][]int |
| 17 | dependents [][]int |
| 18 | // reachable[i] holds every index that transitively depends on i, so the |
| 19 | // preflight can tell ordered items from genuinely concurrent ones. |
| 20 | reachable []map[int]bool |
| 21 | failFast bool |
| 22 | } |
| 23 | |
| 24 | // newFleetPlan validates ids and edges before anything runs. An unknown id, a |
| 25 | // duplicate, a self-edge, or a cycle fails the whole call: a fleet that starts |
| 26 | // and then discovers it cannot finish has already spent tokens. |
| 27 | func newFleetPlan(items []fleetTaskItem, failFast bool) (fleetPlan, error) { |
| 28 | n := len(items) |
| 29 | plan := fleetPlan{ |
| 30 | ids: make([]string, n), |
| 31 | deps: make([][]int, n), |
| 32 | dependents: make([][]int, n), |
| 33 | reachable: make([]map[int]bool, n), |
| 34 | failFast: failFast, |
| 35 | } |
| 36 | index := make(map[string]int, n) |
| 37 | for i, item := range items { |
| 38 | id := strings.TrimSpace(item.ID) |
| 39 | if id == "" { |
| 40 | id = strconv.Itoa(i + 1) |
| 41 | } |
| 42 | if prior, dup := index[id]; dup { |
| 43 | return fleetPlan{}, fmt.Errorf("task %d: id %q is already used by task %d", i+1, id, prior+1) |
| 44 | } |
| 45 | index[id] = i |
| 46 | plan.ids[i] = id |
| 47 | } |
| 48 | for i, item := range items { |
| 49 | for _, raw := range item.DependsOn { |
| 50 | dep := strings.TrimSpace(raw) |
| 51 | target, ok := index[dep] |
| 52 | if !ok { |
| 53 | return fleetPlan{}, fmt.Errorf("task %d (%q): depends_on %q matches no task id", i+1, plan.ids[i], dep) |
| 54 | } |
| 55 | if target == i { |
| 56 | return fleetPlan{}, fmt.Errorf("task %d (%q): depends_on itself", i+1, plan.ids[i]) |
| 57 | } |
| 58 | plan.deps[i] = append(plan.deps[i], target) |
| 59 | plan.dependents[target] = append(plan.dependents[target], i) |
| 60 | } |
| 61 | } |
| 62 | if err := plan.rejectCycles(); err != nil { |
| 63 | return fleetPlan{}, err |
| 64 | } |
| 65 | plan.computeReachability() |
| 66 | return plan, nil |
| 67 | } |
| 68 | |
| 69 | // rejectCycles runs Kahn's algorithm; anything left unvisited is in a cycle. |
| 70 | func (p fleetPlan) rejectCycles() error { |
| 71 | pending := p.pendingCounts() |
| 72 | queue := p.roots() |
| 73 | visited := 0 |
| 74 | for len(queue) > 0 { |
| 75 | current := queue[0] |
| 76 | queue = queue[1:] |
| 77 | visited++ |
| 78 | for _, next := range p.dependents[current] { |
| 79 | pending[next]-- |
| 80 | if pending[next] == 0 { |
| 81 | queue = append(queue, next) |
| 82 | } |
| 83 | } |
| 84 | } |
| 85 | if visited == len(p.ids) { |
| 86 | return nil |
| 87 | } |
| 88 | var stuck []string |
| 89 | for i, count := range pending { |
| 90 | if count > 0 { |
| 91 | stuck = append(stuck, p.ids[i]) |
| 92 | } |
| 93 | } |
| 94 | return fmt.Errorf("depends_on forms a cycle through: %s", strings.Join(stuck, ", ")) |
| 95 | } |
| 96 | |
| 97 | func (p fleetPlan) pendingCounts() []int { |
| 98 | out := make([]int, len(p.ids)) |
| 99 | for i := range p.ids { |
| 100 | out[i] = len(p.deps[i]) |
| 101 | } |
| 102 | return out |
| 103 | } |
| 104 | |
| 105 | func (p fleetPlan) roots() []int { |
| 106 | var out []int |
| 107 | for i := range p.ids { |
| 108 | if len(p.deps[i]) == 0 { |
| 109 | out = append(out, i) |
| 110 | } |
| 111 | } |
| 112 | return out |
| 113 | } |
| 114 | |
| 115 | func (p *fleetPlan) computeReachability() { |
| 116 | for i := range p.ids { |
| 117 | seen := map[int]bool{} |
| 118 | var walk func(int) |
| 119 | walk = func(from int) { |
| 120 | for _, next := range p.dependents[from] { |
| 121 | if seen[next] { |
| 122 | continue |
| 123 | } |
| 124 | seen[next] = true |
| 125 | walk(next) |
| 126 | } |
| 127 | } |
| 128 | walk(i) |
| 129 | p.reachable[i] = seen |
| 130 | } |
| 131 | } |
| 132 | |
| 133 | // describe names an item by its caller-visible position, adding the id only |
| 134 | // when the caller chose one, so diagnostics stay readable either way. |
| 135 | func (p fleetPlan) describe(i int) string { |
| 136 | position := strconv.Itoa(i + 1) |
| 137 | if p.ids[i] == position { |
| 138 | return "task " + position |
| 139 | } |
| 140 | return fmt.Sprintf("task %s (%q)", position, p.ids[i]) |
| 141 | } |
| 142 | |
| 143 | // ordered reports whether one of the two items must finish before the other |
| 144 | // starts, in either direction. |
| 145 | func (p fleetPlan) ordered(a, b int) bool { |
| 146 | return p.reachable[a][b] || p.reachable[b][a] |
| 147 | } |
| 148 | |
| 149 | // validateConcurrentWriteClaims rejects overlapping write claims only for items |
| 150 | // that can actually run at the same time. Two writers joined by a dependency are |
| 151 | // serialised by the graph, so an implement → review chain may legitimately share |
| 152 | // paths that two parallel writers never could. |
| 153 | func (p fleetPlan) validateConcurrentWriteClaims(claims []WritePathSet) error { |
| 154 | for i := range claims { |
| 155 | if claims[i].Empty() { |
| 156 | continue |
| 157 | } |
| 158 | for j := i + 1; j < len(claims); j++ { |
| 159 | if claims[j].Empty() || p.ordered(i, j) { |
| 160 | continue |
| 161 | } |
| 162 | if claims[i].WholeWorkspace || claims[j].WholeWorkspace { |
| 163 | continue |
| 164 | } |
| 165 | if ScheduleOverlaps(claims[i], claims[j]) { |
| 166 | return fmt.Errorf("%s and %s can run at the same time and their write claims conflict; add a depends_on between them or give them disjoint write_paths", |
| 167 | p.describe(i), p.describe(j)) |
| 168 | } |
| 169 | } |
| 170 | } |
| 171 | return nil |
| 172 | } |
| 173 | |
| 174 | // skipDependents marks everything downstream of a failed item as skipped. A |
| 175 | // dependent never runs on a broken input: it would burn tokens to produce a |
| 176 | // result the parent must discard. |
| 177 | func (p fleetPlan) skipDependents(results []fleetItemResult, failed int) { |
| 178 | for idx := range p.reachable[failed] { |
| 179 | if results[idx].status != fleetItemPending { |
| 180 | continue |
| 181 | } |
| 182 | results[idx].status = fleetItemSkipped |
| 183 | results[idx].err = fmt.Errorf("skipped: depends on %q, which did not complete", p.ids[failed]) |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | // errFleetBranchNotStarted marks a task whose branch was cut before it ran. |
| 188 | var errFleetBranchNotStarted = fmt.Errorf("skipped: the fleet stopped starting new tasks") |
| 189 | |
| 190 | func firstNonNilErr(errs ...error) error { |
| 191 | for _, err := range errs { |
| 192 | if err != nil { |
| 193 | return err |
| 194 | } |
| 195 | } |
| 196 | return nil |
| 197 | } |
| 198 | |
| 199 | // driveFleet starts items as their dependencies complete and collects every |
| 200 | // terminal result. Started items always publish one, including after |
| 201 | // cancellation, so partial writer work is never reported as a task that never |
| 202 | // ran. It returns whether the run ended without every item completing. |
| 203 | func driveFleet(ctx context.Context, plan fleetPlan, results []fleetItemResult, doneCh <-chan fleetItemResult, wait func(), startOne func(int)) bool { |
| 204 | pending := plan.pendingCounts() |
| 205 | started, completed := 0, 0 |
| 206 | stopStarting := false |
| 207 | launch := func(idx int) { |
| 208 | if stopStarting || ctx.Err() != nil || results[idx].status != fleetItemPending { |
| 209 | return |
| 210 | } |
| 211 | startOne(idx) |
| 212 | started++ |
| 213 | } |
| 214 | for _, idx := range plan.roots() { |
| 215 | launch(idx) |
| 216 | } |
| 217 | |
| 218 | cancelled := false |
| 219 | for completed < started && !cancelled { |
| 220 | select { |
| 221 | case r := <-doneCh: |
| 222 | results[r.index] = r |
| 223 | completed++ |
| 224 | if r.status != fleetItemCompleted { |
| 225 | plan.skipDependents(results, r.index) |
| 226 | if plan.failFast { |
| 227 | stopStarting = true |
| 228 | } |
| 229 | continue |
| 230 | } |
| 231 | for _, next := range plan.dependents[r.index] { |
| 232 | if pending[next]--; pending[next] == 0 { |
| 233 | launch(next) |
| 234 | } |
| 235 | } |
| 236 | case <-ctx.Done(): |
| 237 | cancelled = true |
| 238 | } |
| 239 | } |
| 240 | // doneCh is buffered for every item, so workers can always publish while |
| 241 | // this goroutine waits; drain the outstanding ones rather than overwriting |
| 242 | // their real status with skipped. |
| 243 | wait() |
| 244 | for completed < started { |
| 245 | r := <-doneCh |
| 246 | results[r.index] = r |
| 247 | completed++ |
| 248 | } |
| 249 | for i := range results { |
| 250 | if results[i].status != fleetItemPending { |
| 251 | continue |
| 252 | } |
| 253 | results[i].status = fleetItemSkipped |
| 254 | if results[i].err == nil { |
| 255 | results[i].err = firstNonNilErr(ctx.Err(), errFleetBranchNotStarted) |
| 256 | } |
| 257 | } |
| 258 | return cancelled |
| 259 | } |
| 260 |