mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
feat(agent): make fleet a small dependency graph
A fleet item may now declare id and depends_on. That is the entire graph vocabulary — no conditions, no expressions, no dynamic fan-out — and it is enough for research → implement backend/frontend → integration → review. Dependencies live on the plan, never on a task spec. ProfileExecSpec is unchanged, which is what keeps depends_on from becoming the first keyword of a workflow language: what a task is does not depend on what ran before it. Everything is validated before anything starts, alongside the existing write claim preflight — duplicate ids, ids no task declares, self-edges, and cycles all fail the whole call, because a fleet that cannot finish has already spent tokens by the time it discovers that. The graph also relaxes the write-claim preflight where it should: only items that can actually run at the same time need disjoint write_paths. An implement → review pair is serialised by its edge and may share paths, which a flat fleet had no way to express. Failure handling stays at one knob. A failed or skipped task always skips its whole downstream branch — a dependent on a broken input only produces a result the parent must discard. Independent branches keep going unless fail_fast is set, which stops starting new tasks while leaving running ones to finish, so a writer is never abandoned mid-write. Cache-impact: the fleet schema and description are in the parent prefix, so this is a one-time invalidation. The regenerated golden baseline puts it at +273 tool-schema tokens (13801 → 14074); SystemHash is unchanged.
This commit is contained in:
1 parent
85cacad3b1
commit
3b9dff68de
8 files changed
+554
-50
No files matched your search
@@ -735,6 +735,36 @@ the record, and two runs that behaved differently can be diffed instead of
|
||||
guessed at. The capsule holds references and digests only — never copied parent
|
||||
context, which is what keeps delegation cheap and the child prefix cacheable.
|
||||
|
||||
### 3.14 Fleet is a small dependency graph
|
||||
|
||||
A fleet item may declare `id` and `depends_on`. That is the whole graph
|
||||
vocabulary: no conditions, no expressions, no dynamic fan-out. It is enough for
|
||||
|
||||
```
|
||||
research ──▶ implement backend ──┐
|
||||
└──▶ implement frontend ─┴──▶ integration test ──▶ review
|
||||
```
|
||||
|
||||
Ids default to the 1-based position. A duplicate id, an id no task declares, a
|
||||
self-edge, or a cycle fails preflight, so a fleet that cannot finish never
|
||||
starts. Items run as soon as their dependencies complete; items with no ordering
|
||||
between them run in parallel under the same session scheduler as before.
|
||||
|
||||
Dependencies are a property of the graph, never of a task: they live in the
|
||||
fleet plan and never reach `ProfileExecSpec`, which is what keeps `depends_on`
|
||||
from becoming the first keyword of a workflow language.
|
||||
|
||||
The graph relaxes the write-claim preflight in the one place it should. Only
|
||||
items that can run at the same time need disjoint `write_paths`; an
|
||||
`implement → review` pair is serialised by its edge and may share paths, which
|
||||
a flat fleet could not express.
|
||||
|
||||
Failure handling has one knob. A failed or skipped task always skips its whole
|
||||
downstream branch — running a dependent on a broken input only buys a result the
|
||||
parent must discard. Independent branches keep going unless `fail_fast` is set,
|
||||
which stops *starting* new tasks; tasks already running are left to finish so a
|
||||
writer is never abandoned mid-write.
|
||||
|
||||
## 4. Data Types (`internal/provider`)
|
||||
|
||||
```go
|
||||
|
||||
@@ -278,6 +278,23 @@ Profile 描述的是 worker,不是一次运行。委派由五个彼此独立
|
||||
|
||||
每次运行都会在其 transcript sidecar 中记录一份 `ContextCapsule`:workspace、系统提示来源与哈希、解析后的工具范围与 schema 哈希、model 与 effort、父会话与父工具调用 id、续接的 transcript,以及一个所有字段均为 false 的 `inherited` 块。`capsuleHash` 是它的稳定标识,因此"为什么这个 reviewer 没看到那条约束"可以从记录回答,两次行为不同的运行也可以直接比对而不是猜。capsule 只保存引用与摘要——绝不复制父上下文,这正是委派保持低成本、子前缀保持可缓存的原因。
|
||||
|
||||
### 3.14 fleet 是一张小依赖图
|
||||
|
||||
fleet item 可以声明 `id` 与 `depends_on`。图的词汇就这么多:没有条件、没有表达式、没有动态扩散。它足以表达
|
||||
|
||||
```
|
||||
research ──▶ implement backend ──┐
|
||||
└──▶ implement frontend ─┴──▶ integration test ──▶ review
|
||||
```
|
||||
|
||||
id 默认取 1 起的序号。重复 id、指向不存在任务的 id、自环、成环都会在 preflight 失败——一个注定跑不完的 fleet 绝不会开始。依赖完成后条目立即启动;彼此无序的条目仍按既有 session scheduler 并发。
|
||||
|
||||
依赖是图的性质,不是任务的性质:它只存在于 fleet plan 中,绝不进入 `ProfileExecSpec`。这正是让 `depends_on` 不至于成为某种 workflow 语言第一个关键字的原因。
|
||||
|
||||
图恰好在该放松的地方放松了写声明 preflight:只有**可能同时运行**的条目才需要互不重叠的 `write_paths`;`implement → review` 这对被边串行化,可以共享路径——扁平 fleet 无法表达这一点。
|
||||
|
||||
失败处理只有一个开关。失败或被跳过的任务永远会跳过其整条下游分支——在坏输入上跑依赖项,只会换来父智能体必须丢弃的结果。除非设置 `fail_fast`,独立分支继续推进;`fail_fast` 停止的是**启动**新任务,已在运行的任务留待自然结束,因此写入者绝不会被中途丢弃。
|
||||
|
||||
## 4. 数据类型
|
||||
|
||||
provider 层的核心类型包括 `Role`、`Message`、`ToolCall`、`ToolSchema`、`Request` 和 streaming `Chunk`。`Message` 保留 `tool_calls`、`tool_call_id` 与 `name`;`Chunk` 区分 text、tool call、done 和 error。字段定义以英文规范及 `internal/provider` 源码为准。
|
||||
|
||||
+17
-43
@@ -36,7 +36,7 @@ func NewFleetTool(taskTool *TaskTool) *FleetTool {
|
||||
func (*FleetTool) Name() string { return "fleet" }
|
||||
|
||||
func (*FleetTool) Description() string {
|
||||
return "Dispatch 2–64 sub-agent tasks in parallel and return bounded previews plus stable Subagent references for full-result retrieval from completed persisted children with read_subagent_result. Each item may select a profile, model, effort, tools, write_paths, or read_only. Multiple writers must declare non-overlapping write_paths; omitted write_paths claim the whole workspace, so two or more writers without paths fail preflight before any task starts. Independent failure is the default: one failure does not cancel others. Background mode returns a fleet job id collectable with wait."
|
||||
return "Dispatch 2–64 sub-agent tasks as a small dependency graph and return bounded previews plus stable Subagent references for full-result retrieval from completed persisted children with read_subagent_result. Each item may select a profile, model, effort, tools, write_paths, or read_only, and may declare depends_on to run after other items (research → implement → review). Tasks with no dependency between them run in parallel and must declare non-overlapping write_paths; ordered tasks may share paths. Omitted write_paths claim the whole workspace, so two or more concurrent writers without paths fail preflight before any task starts. A failed task's dependents are skipped; independent branches keep going unless fail_fast is set. Background mode returns a fleet job id collectable with wait."
|
||||
}
|
||||
|
||||
func (*FleetTool) Schema() json.RawMessage {
|
||||
@@ -52,9 +52,11 @@ func (*FleetTool) Schema() json.RawMessage {
|
||||
"type":"object",
|
||||
"properties":{
|
||||
"prompt":{"type":"string","description":"Task prompt for the sub-agent."},
|
||||
"id":{"type":"string","description":"Optional stable id for this task, referenced by other tasks' depends_on. Defaults to the 1-based position."},
|
||||
"depends_on":{"type":"array","items":{"type":"string"},"description":"Ids of tasks that must complete before this one starts. Unknown ids, self-edges, and cycles fail preflight. A task whose dependency fails or is skipped is skipped too. Ordered tasks may share write_paths; only tasks that can run at the same time need disjoint claims."},
|
||||
"description":{"type":"string","description":"Optional short label shown in the job list."},
|
||||
"profile":{"type":"string","description":"Optional runAs=subagent profile name."},
|
||||
"write_paths":{"type":"array","items":{"type":"string"},"description":"Write targets for this item. Parallel writers must declare non-overlapping paths. Omitting write_paths claims the whole workspace; multiple whole-workspace claims (or any path overlap) fail preflight and start nothing."},
|
||||
"write_paths":{"type":"array","items":{"type":"string"},"description":"Write targets for this item. Writers that can run at the same time must declare non-overlapping paths; writers ordered by depends_on may share them. Omitting write_paths claims the whole workspace; two concurrent whole-workspace claims (or any overlap between concurrent writers) fail preflight and start nothing."},
|
||||
"read_only":{"type":"boolean","description":"Force the read-only registry even if the profile is writable."},
|
||||
"tools":{"type":"array","items":{"type":"string"},"description":"Optional tool whitelist (intersected with profile allowed-tools)."},
|
||||
"max_steps":{"type":"integer","description":"Optional max tool-call rounds.","minimum":1},
|
||||
@@ -64,6 +66,7 @@ func (*FleetTool) Schema() json.RawMessage {
|
||||
"required":["prompt"]
|
||||
}
|
||||
},
|
||||
"fail_fast":{"type":"boolean","description":"Stop starting new tasks after the first failure. Tasks already running are left to finish so partial writes are not abandoned mid-flight. Omitted (the default) means independent branches keep going; a failed task's dependents are skipped either way."},
|
||||
"run_in_background":{"type":"boolean","description":"Run the whole fleet asynchronously and return a job id collectable with wait. Items queue for concurrency/write slots inside the job."}
|
||||
},
|
||||
"required":["tasks"]
|
||||
@@ -74,6 +77,8 @@ func (*FleetTool) ReadOnly() bool { return false }
|
||||
|
||||
type fleetTaskItem struct {
|
||||
Prompt string `json:"prompt"`
|
||||
ID string `json:"id"`
|
||||
DependsOn []string `json:"depends_on"`
|
||||
Description string `json:"description"`
|
||||
Profile string `json:"profile"`
|
||||
WritePaths []string `json:"write_paths"`
|
||||
@@ -157,6 +162,7 @@ func (f *FleetTool) Execute(ctx context.Context, args json.RawMessage) (result s
|
||||
|
||||
var params struct {
|
||||
Tasks []fleetTaskItem `json:"tasks"`
|
||||
FailFast bool `json:"fail_fast"`
|
||||
RunInBackground bool `json:"run_in_background"`
|
||||
}
|
||||
dec := json.NewDecoder(bytes.NewReader(args))
|
||||
@@ -200,7 +206,11 @@ func (f *FleetTool) Execute(ctx context.Context, args json.RawMessage) (result s
|
||||
claims[i] = spec.Grant.WritePaths
|
||||
}
|
||||
}
|
||||
if err := ValidateNonOverlappingWriteClaims(claims); err != nil {
|
||||
plan, err := newFleetPlan(params.Tasks, params.FailFast)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("fleet preflight: %w", err)
|
||||
}
|
||||
if err := plan.validateConcurrentWriteClaims(claims); err != nil {
|
||||
return "", fmt.Errorf("fleet preflight: %w", err)
|
||||
}
|
||||
|
||||
@@ -248,7 +258,7 @@ func (f *FleetTool) Execute(ctx context.Context, args json.RawMessage) (result s
|
||||
// The job shares the Execute-level merger so the group lifecycle
|
||||
// events and the child previews ride the same pacing budget.
|
||||
jobCtx = withSubagentProgressMerger(jobCtx, merger)
|
||||
return f.runFleet(jobCtx, groupSink, specs, parentID)
|
||||
return f.runFleet(jobCtx, groupSink, specs, plan, parentID)
|
||||
})
|
||||
// runFleet (inside the job) owns the terminal and merger close from
|
||||
// here on. Foreground runFleet hands off only the terminal; Execute
|
||||
@@ -259,10 +269,10 @@ func (f *FleetTool) Execute(ctx context.Context, args json.RawMessage) (result s
|
||||
}
|
||||
|
||||
lifecycleHandoff = true
|
||||
return f.runFleet(ctx, groupSink, specs, groupParentID)
|
||||
return f.runFleet(ctx, groupSink, specs, plan, groupParentID)
|
||||
}
|
||||
|
||||
func (f *FleetTool) runFleet(ctx context.Context, sink event.Sink, specs []ProfileExecSpec, groupParentID string) (result string, err error) {
|
||||
func (f *FleetTool) runFleet(ctx context.Context, sink event.Sink, specs []ProfileExecSpec, plan fleetPlan, groupParentID string) (result string, err error) {
|
||||
if sink == nil {
|
||||
sink = event.Discard
|
||||
}
|
||||
@@ -351,43 +361,7 @@ func (f *FleetTool) runFleet(ctx context.Context, sink event.Sink, specs []Profi
|
||||
})
|
||||
}
|
||||
|
||||
// Mark only genuinely unstarted items skipped. Started items always publish
|
||||
// a terminal result, including after cancellation, so partial writer work is
|
||||
// never misreported as a task that did not run.
|
||||
started := 0
|
||||
for i := range specs {
|
||||
if ctx.Err() != nil {
|
||||
for j := i; j < n; j++ {
|
||||
results[j].status = fleetItemSkipped
|
||||
results[j].err = ctx.Err()
|
||||
}
|
||||
break
|
||||
}
|
||||
startOne(i)
|
||||
started++
|
||||
}
|
||||
|
||||
completed := 0
|
||||
cancelled := false
|
||||
for completed < started && !cancelled {
|
||||
select {
|
||||
case r := <-doneCh:
|
||||
results[r.index] = r
|
||||
completed++
|
||||
case <-ctx.Done():
|
||||
cancelled = true
|
||||
}
|
||||
}
|
||||
// doneCh is buffered for every item, so workers can always publish their
|
||||
// terminal result while this goroutine waits. Once they stop, drain exactly
|
||||
// the outstanding started items and preserve their real completed/cancelled
|
||||
// status instead of replacing it with skipped.
|
||||
wg.Wait()
|
||||
for completed < started {
|
||||
r := <-doneCh
|
||||
results[r.index] = r
|
||||
completed++
|
||||
}
|
||||
cancelled := driveFleet(ctx, plan, results, doneCh, wg.Wait, startOne)
|
||||
for _, r := range results {
|
||||
if r.status == fleetItemCancelled || r.status == fleetItemSkipped {
|
||||
cancelled = true
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// fleetPlan is the validated dependency graph for one fleet call. Dependencies
|
||||
// live here, on the graph, and never on a task spec: what a task is does not
|
||||
// depend on what ran before it. Keeping them apart is what stops fleet from
|
||||
// growing into a workflow language.
|
||||
type fleetPlan struct {
|
||||
ids []string
|
||||
deps [][]int
|
||||
dependents [][]int
|
||||
// reachable[i] holds every index that transitively depends on i, so the
|
||||
// preflight can tell ordered items from genuinely concurrent ones.
|
||||
reachable []map[int]bool
|
||||
failFast bool
|
||||
}
|
||||
|
||||
// newFleetPlan validates ids and edges before anything runs. An unknown id, a
|
||||
// duplicate, a self-edge, or a cycle fails the whole call: a fleet that starts
|
||||
// and then discovers it cannot finish has already spent tokens.
|
||||
func newFleetPlan(items []fleetTaskItem, failFast bool) (fleetPlan, error) {
|
||||
n := len(items)
|
||||
plan := fleetPlan{
|
||||
ids: make([]string, n),
|
||||
deps: make([][]int, n),
|
||||
dependents: make([][]int, n),
|
||||
reachable: make([]map[int]bool, n),
|
||||
failFast: failFast,
|
||||
}
|
||||
index := make(map[string]int, n)
|
||||
for i, item := range items {
|
||||
id := strings.TrimSpace(item.ID)
|
||||
if id == "" {
|
||||
id = strconv.Itoa(i + 1)
|
||||
}
|
||||
if prior, dup := index[id]; dup {
|
||||
return fleetPlan{}, fmt.Errorf("task %d: id %q is already used by task %d", i+1, id, prior+1)
|
||||
}
|
||||
index[id] = i
|
||||
plan.ids[i] = id
|
||||
}
|
||||
for i, item := range items {
|
||||
for _, raw := range item.DependsOn {
|
||||
dep := strings.TrimSpace(raw)
|
||||
target, ok := index[dep]
|
||||
if !ok {
|
||||
return fleetPlan{}, fmt.Errorf("task %d (%q): depends_on %q matches no task id", i+1, plan.ids[i], dep)
|
||||
}
|
||||
if target == i {
|
||||
return fleetPlan{}, fmt.Errorf("task %d (%q): depends_on itself", i+1, plan.ids[i])
|
||||
}
|
||||
plan.deps[i] = append(plan.deps[i], target)
|
||||
plan.dependents[target] = append(plan.dependents[target], i)
|
||||
}
|
||||
}
|
||||
if err := plan.rejectCycles(); err != nil {
|
||||
return fleetPlan{}, err
|
||||
}
|
||||
plan.computeReachability()
|
||||
return plan, nil
|
||||
}
|
||||
|
||||
// rejectCycles runs Kahn's algorithm; anything left unvisited is in a cycle.
|
||||
func (p fleetPlan) rejectCycles() error {
|
||||
pending := p.pendingCounts()
|
||||
queue := p.roots()
|
||||
visited := 0
|
||||
for len(queue) > 0 {
|
||||
current := queue[0]
|
||||
queue = queue[1:]
|
||||
visited++
|
||||
for _, next := range p.dependents[current] {
|
||||
pending[next]--
|
||||
if pending[next] == 0 {
|
||||
queue = append(queue, next)
|
||||
}
|
||||
}
|
||||
}
|
||||
if visited == len(p.ids) {
|
||||
return nil
|
||||
}
|
||||
var stuck []string
|
||||
for i, count := range pending {
|
||||
if count > 0 {
|
||||
stuck = append(stuck, p.ids[i])
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("depends_on forms a cycle through: %s", strings.Join(stuck, ", "))
|
||||
}
|
||||
|
||||
func (p fleetPlan) pendingCounts() []int {
|
||||
out := make([]int, len(p.ids))
|
||||
for i := range p.ids {
|
||||
out[i] = len(p.deps[i])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (p fleetPlan) roots() []int {
|
||||
var out []int
|
||||
for i := range p.ids {
|
||||
if len(p.deps[i]) == 0 {
|
||||
out = append(out, i)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (p *fleetPlan) computeReachability() {
|
||||
for i := range p.ids {
|
||||
seen := map[int]bool{}
|
||||
var walk func(int)
|
||||
walk = func(from int) {
|
||||
for _, next := range p.dependents[from] {
|
||||
if seen[next] {
|
||||
continue
|
||||
}
|
||||
seen[next] = true
|
||||
walk(next)
|
||||
}
|
||||
}
|
||||
walk(i)
|
||||
p.reachable[i] = seen
|
||||
}
|
||||
}
|
||||
|
||||
// describe names an item by its caller-visible position, adding the id only
|
||||
// when the caller chose one, so diagnostics stay readable either way.
|
||||
func (p fleetPlan) describe(i int) string {
|
||||
position := strconv.Itoa(i + 1)
|
||||
if p.ids[i] == position {
|
||||
return "task " + position
|
||||
}
|
||||
return fmt.Sprintf("task %s (%q)", position, p.ids[i])
|
||||
}
|
||||
|
||||
// ordered reports whether one of the two items must finish before the other
|
||||
// starts, in either direction.
|
||||
func (p fleetPlan) ordered(a, b int) bool {
|
||||
return p.reachable[a][b] || p.reachable[b][a]
|
||||
}
|
||||
|
||||
// validateConcurrentWriteClaims rejects overlapping write claims only for items
|
||||
// that can actually run at the same time. Two writers joined by a dependency are
|
||||
// serialised by the graph, so an implement → review chain may legitimately share
|
||||
// paths that two parallel writers never could.
|
||||
func (p fleetPlan) validateConcurrentWriteClaims(claims []WritePathSet) error {
|
||||
for i := range claims {
|
||||
if claims[i].Empty() {
|
||||
continue
|
||||
}
|
||||
for j := i + 1; j < len(claims); j++ {
|
||||
if claims[j].Empty() || p.ordered(i, j) {
|
||||
continue
|
||||
}
|
||||
if claims[i].Overlaps(claims[j]) {
|
||||
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",
|
||||
p.describe(i), p.describe(j))
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// skipDependents marks everything downstream of a failed item as skipped. A
|
||||
// dependent never runs on a broken input: it would burn tokens to produce a
|
||||
// result the parent must discard.
|
||||
func (p fleetPlan) skipDependents(results []fleetItemResult, failed int) {
|
||||
for idx := range p.reachable[failed] {
|
||||
if results[idx].status != fleetItemPending {
|
||||
continue
|
||||
}
|
||||
results[idx].status = fleetItemSkipped
|
||||
results[idx].err = fmt.Errorf("skipped: depends on %q, which did not complete", p.ids[failed])
|
||||
}
|
||||
}
|
||||
|
||||
// errFleetBranchNotStarted marks a task whose branch was cut before it ran.
|
||||
var errFleetBranchNotStarted = fmt.Errorf("skipped: the fleet stopped starting new tasks")
|
||||
|
||||
func firstNonNilErr(errs ...error) error {
|
||||
for _, err := range errs {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// driveFleet starts items as their dependencies complete and collects every
|
||||
// terminal result. Started items always publish one, including after
|
||||
// cancellation, so partial writer work is never reported as a task that never
|
||||
// ran. It returns whether the run ended without every item completing.
|
||||
func driveFleet(ctx context.Context, plan fleetPlan, results []fleetItemResult, doneCh <-chan fleetItemResult, wait func(), startOne func(int)) bool {
|
||||
pending := plan.pendingCounts()
|
||||
started, completed := 0, 0
|
||||
stopStarting := false
|
||||
launch := func(idx int) {
|
||||
if stopStarting || ctx.Err() != nil || results[idx].status != fleetItemPending {
|
||||
return
|
||||
}
|
||||
startOne(idx)
|
||||
started++
|
||||
}
|
||||
for _, idx := range plan.roots() {
|
||||
launch(idx)
|
||||
}
|
||||
|
||||
cancelled := false
|
||||
for completed < started && !cancelled {
|
||||
select {
|
||||
case r := <-doneCh:
|
||||
results[r.index] = r
|
||||
completed++
|
||||
if r.status != fleetItemCompleted {
|
||||
plan.skipDependents(results, r.index)
|
||||
if plan.failFast {
|
||||
stopStarting = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
for _, next := range plan.dependents[r.index] {
|
||||
if pending[next]--; pending[next] == 0 {
|
||||
launch(next)
|
||||
}
|
||||
}
|
||||
case <-ctx.Done():
|
||||
cancelled = true
|
||||
}
|
||||
}
|
||||
// doneCh is buffered for every item, so workers can always publish while
|
||||
// this goroutine waits; drain the outstanding ones rather than overwriting
|
||||
// their real status with skipped.
|
||||
wait()
|
||||
for completed < started {
|
||||
r := <-doneCh
|
||||
results[r.index] = r
|
||||
completed++
|
||||
}
|
||||
for i := range results {
|
||||
if results[i].status != fleetItemPending {
|
||||
continue
|
||||
}
|
||||
results[i].status = fleetItemSkipped
|
||||
if results[i].err == nil {
|
||||
results[i].err = firstNonNilErr(ctx.Err(), errFleetBranchNotStarted)
|
||||
}
|
||||
}
|
||||
return cancelled
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/event"
|
||||
"reasonix/internal/provider"
|
||||
"reasonix/internal/tool"
|
||||
)
|
||||
|
||||
func planFor(t *testing.T, items ...fleetTaskItem) (fleetPlan, error) {
|
||||
t.Helper()
|
||||
return newFleetPlan(items, false)
|
||||
}
|
||||
|
||||
func TestFleetPlanRejectsBrokenGraphsBeforeAnythingRuns(t *testing.T) {
|
||||
for name, tc := range map[string]struct {
|
||||
items []fleetTaskItem
|
||||
want string
|
||||
}{
|
||||
"duplicate id": {
|
||||
items: []fleetTaskItem{{ID: "a"}, {ID: "a"}},
|
||||
want: "already used",
|
||||
},
|
||||
"unknown dependency": {
|
||||
items: []fleetTaskItem{{ID: "a"}, {ID: "b", DependsOn: []string{"nope"}}},
|
||||
want: "matches no task id",
|
||||
},
|
||||
"self dependency": {
|
||||
items: []fleetTaskItem{{ID: "a", DependsOn: []string{"a"}}, {ID: "b"}},
|
||||
want: "depends_on itself",
|
||||
},
|
||||
"cycle": {
|
||||
items: []fleetTaskItem{
|
||||
{ID: "a", DependsOn: []string{"c"}},
|
||||
{ID: "b", DependsOn: []string{"a"}},
|
||||
{ID: "c", DependsOn: []string{"b"}},
|
||||
},
|
||||
want: "cycle",
|
||||
},
|
||||
} {
|
||||
if _, err := planFor(t, tc.items...); err == nil || !strings.Contains(err.Error(), tc.want) {
|
||||
t.Errorf("%s: err = %v, want one mentioning %q", name, err, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFleetPlanOrdersTransitiveDependents(t *testing.T) {
|
||||
plan, err := planFor(t,
|
||||
fleetTaskItem{ID: "research"},
|
||||
fleetTaskItem{ID: "backend", DependsOn: []string{"research"}},
|
||||
fleetTaskItem{ID: "frontend", DependsOn: []string{"research"}},
|
||||
fleetTaskItem{ID: "integration", DependsOn: []string{"backend", "frontend"}},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !plan.ordered(0, 3) {
|
||||
t.Error("integration transitively depends on research and must be ordered against it")
|
||||
}
|
||||
if plan.ordered(1, 2) {
|
||||
t.Error("backend and frontend share a dependency but not an order: they run in parallel")
|
||||
}
|
||||
if roots := plan.roots(); len(roots) != 1 || roots[0] != 0 {
|
||||
t.Errorf("roots = %v, want just research", roots)
|
||||
}
|
||||
}
|
||||
|
||||
// The unlock: implement → review legitimately touch the same files, which a
|
||||
// flat fleet could never express.
|
||||
func TestFleetOrderedWritersMayShareWritePaths(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
claim, err := NormalizeWritePaths(root, []string{"api"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ordered, err := planFor(t,
|
||||
fleetTaskItem{ID: "implement"},
|
||||
fleetTaskItem{ID: "review", DependsOn: []string{"implement"}},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ordered.validateConcurrentWriteClaims([]WritePathSet{claim, claim}); err != nil {
|
||||
t.Fatalf("ordered writers must be allowed to share paths: %v", err)
|
||||
}
|
||||
|
||||
concurrent, err := planFor(t, fleetTaskItem{ID: "a"}, fleetTaskItem{ID: "b"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := concurrent.validateConcurrentWriteClaims([]WritePathSet{claim, claim}); err == nil {
|
||||
t.Fatal("two writers that can run at once must still fail preflight on overlap")
|
||||
}
|
||||
}
|
||||
|
||||
func TestFleetPlanSkipsWholeDownstreamBranch(t *testing.T) {
|
||||
plan, err := planFor(t,
|
||||
fleetTaskItem{ID: "research"},
|
||||
fleetTaskItem{ID: "implement", DependsOn: []string{"research"}},
|
||||
fleetTaskItem{ID: "review", DependsOn: []string{"implement"}},
|
||||
fleetTaskItem{ID: "unrelated"},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
results := make([]fleetItemResult, 4)
|
||||
for i := range results {
|
||||
results[i] = fleetItemResult{index: i, status: fleetItemPending}
|
||||
}
|
||||
results[0].status = fleetItemFailed
|
||||
plan.skipDependents(results, 0)
|
||||
|
||||
if results[1].status != fleetItemSkipped || results[2].status != fleetItemSkipped {
|
||||
t.Fatalf("statuses = %q/%q, want the whole downstream branch skipped", results[1].status, results[2].status)
|
||||
}
|
||||
if !strings.Contains(results[2].err.Error(), `depends on "research"`) {
|
||||
t.Fatalf("skip reason = %v, want it to name the broken dependency", results[2].err)
|
||||
}
|
||||
if results[3].status != fleetItemPending {
|
||||
t.Error("an unrelated branch must not be skipped by another branch's failure")
|
||||
}
|
||||
}
|
||||
|
||||
// End to end: a dependent never runs when its dependency failed.
|
||||
func TestFleetSkipsDependentsOfFailedTask(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
prov := &fleetScriptedFailureProvider{}
|
||||
reg := tool.NewRegistry()
|
||||
reg.Add(fakeReadFileTool{})
|
||||
task := NewTaskTool(prov, nil, reg, 20, 0, 0, 0, 0, 0, 0, 0.0, "", "sys", nil, 0, "", "", nil).
|
||||
WithTranscripts(mustSubagentStore(t), root, "base", "high").
|
||||
WithScheduler(NewSubagentScheduler(4, 4))
|
||||
fleet := NewFleetTool(task)
|
||||
ctx := withCallContext(context.Background(), "fleet-call", event.Discard, nil, false)
|
||||
|
||||
out, _ := fleet.Execute(ctx, json.RawMessage(`{"tasks":[
|
||||
{"id":"research","prompt":"FAIL research","read_only":true},
|
||||
{"id":"implement","prompt":"implement","depends_on":["research"],"write_paths":["api"]},
|
||||
{"id":"sibling","prompt":"sibling","read_only":true}
|
||||
]}`))
|
||||
|
||||
if !strings.Contains(out, "skipped") || !strings.Contains(out, `depends on "research"`) {
|
||||
t.Fatalf("aggregate must report the dependent as skipped with its reason:\n%s", out)
|
||||
}
|
||||
if prov.ran("implement") {
|
||||
t.Fatal("a dependent of a failed task must never start")
|
||||
}
|
||||
if !prov.ran("sibling") {
|
||||
t.Fatal("an independent branch must still run when another branch fails")
|
||||
}
|
||||
}
|
||||
|
||||
type fleetScriptedFailureProvider struct {
|
||||
mu sync.Mutex
|
||||
seen map[string]bool
|
||||
}
|
||||
|
||||
func (p *fleetScriptedFailureProvider) Name() string { return "fleet-failure" }
|
||||
|
||||
func (p *fleetScriptedFailureProvider) ran(name string) bool {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.seen[name]
|
||||
}
|
||||
|
||||
func (p *fleetScriptedFailureProvider) Stream(_ context.Context, req provider.Request) (<-chan provider.Chunk, error) {
|
||||
last := ""
|
||||
for _, m := range req.Messages {
|
||||
if m.Role == provider.RoleUser {
|
||||
last = m.Content
|
||||
}
|
||||
}
|
||||
p.mu.Lock()
|
||||
if p.seen == nil {
|
||||
p.seen = map[string]bool{}
|
||||
}
|
||||
for _, name := range []string{"implement", "sibling"} {
|
||||
if strings.Contains(last, name) {
|
||||
p.seen[name] = true
|
||||
}
|
||||
}
|
||||
p.mu.Unlock()
|
||||
if strings.Contains(last, "FAIL") {
|
||||
return nil, errors.New("scripted provider failure")
|
||||
}
|
||||
ch := make(chan provider.Chunk, 2)
|
||||
ch <- provider.Chunk{Type: provider.ChunkText, Text: "done"}
|
||||
ch <- provider.Chunk{Type: provider.ChunkDone}
|
||||
close(ch)
|
||||
return ch, nil
|
||||
}
|
||||
+3
-3
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"SystemHash": "c4a54b2f61c5e15f",
|
||||
"ToolsHash": "ab2f66c27b039773",
|
||||
"PrefixHash": "9e132ad357f9874b",
|
||||
"ToolsHash": "5d9946fc2cda6d80",
|
||||
"PrefixHash": "18c81fd05e7f17b8",
|
||||
"LogRewriteVersion": 0,
|
||||
"ToolSchemaTokens": 13801
|
||||
"ToolSchemaTokens": 14074
|
||||
}
|
||||
+17
-2
@@ -442,9 +442,13 @@
|
||||
},
|
||||
{
|
||||
"name": "fleet",
|
||||
"description": "Dispatch 2–64 sub-agent tasks in parallel and return bounded previews plus stable Subagent references for full-result retrieval from completed persisted children with read_subagent_result. Each item may select a profile, model, effort, tools, write_paths, or read_only. Multiple writers must declare non-overlapping write_paths; omitted write_paths claim the whole workspace, so two or more writers without paths fail preflight before any task starts. Independent failure is the default: one failure does not cancel others. Background mode returns a fleet job id collectable with wait.",
|
||||
"description": "Dispatch 2–64 sub-agent tasks as a small dependency graph and return bounded previews plus stable Subagent references for full-result retrieval from completed persisted children with read_subagent_result. Each item may select a profile, model, effort, tools, write_paths, or read_only, and may declare depends_on to run after other items (research → implement → review). Tasks with no dependency between them run in parallel and must declare non-overlapping write_paths; ordered tasks may share paths. Omitted write_paths claim the whole workspace, so two or more concurrent writers without paths fail preflight before any task starts. A failed task's dependents are skipped; independent branches keep going unless fail_fast is set. Background mode returns a fleet job id collectable with wait.",
|
||||
"parameters": {
|
||||
"properties": {
|
||||
"fail_fast": {
|
||||
"description": "Stop starting new tasks after the first failure. Tasks already running are left to finish so partial writes are not abandoned mid-flight. Omitted (the default) means independent branches keep going; a failed task's dependents are skipped either way.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"run_in_background": {
|
||||
"description": "Run the whole fleet asynchronously and return a job id collectable with wait. Items queue for concurrency/write slots inside the job.",
|
||||
"type": "boolean"
|
||||
@@ -453,6 +457,13 @@
|
||||
"description": "Array of 2–64 sub-tasks to run under the session scheduler.",
|
||||
"items": {
|
||||
"properties": {
|
||||
"depends_on": {
|
||||
"description": "Ids of tasks that must complete before this one starts. Unknown ids, self-edges, and cycles fail preflight. A task whose dependency fails or is skipped is skipped too. Ordered tasks may share write_paths; only tasks that can run at the same time need disjoint claims.",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"type": "array"
|
||||
},
|
||||
"description": {
|
||||
"description": "Optional short label shown in the job list.",
|
||||
"type": "string"
|
||||
@@ -461,6 +472,10 @@
|
||||
"description": "Optional reasoning effort override.",
|
||||
"type": "string"
|
||||
},
|
||||
"id": {
|
||||
"description": "Optional stable id for this task, referenced by other tasks' depends_on. Defaults to the 1-based position.",
|
||||
"type": "string"
|
||||
},
|
||||
"max_steps": {
|
||||
"description": "Optional max tool-call rounds.",
|
||||
"minimum": 1,
|
||||
@@ -490,7 +505,7 @@
|
||||
"type": "array"
|
||||
},
|
||||
"write_paths": {
|
||||
"description": "Write targets for this item. Parallel writers must declare non-overlapping paths. Omitting write_paths claims the whole workspace; multiple whole-workspace claims (or any path overlap) fail preflight and start nothing.",
|
||||
"description": "Write targets for this item. Writers that can run at the same time must declare non-overlapping paths; writers ordered by depends_on may share them. Omitting write_paths claims the whole workspace; two concurrent whole-workspace claims (or any overlap between concurrent writers) fail preflight and start nothing.",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+17
-2
@@ -423,10 +423,14 @@
|
||||
},
|
||||
{
|
||||
"Name": "fleet",
|
||||
"Description": "Dispatch 2–64 sub-agent tasks in parallel and return bounded previews plus stable Subagent references for full-result retrieval from completed persisted children with read_subagent_result. Each item may select a profile, model, effort, tools, write_paths, or read_only. Multiple writers must declare non-overlapping write_paths; omitted write_paths claim the whole workspace, so two or more writers without paths fail preflight before any task starts. Independent failure is the default: one failure does not cancel others. Background mode returns a fleet job id collectable with wait.",
|
||||
"Description": "Dispatch 2–64 sub-agent tasks as a small dependency graph and return bounded previews plus stable Subagent references for full-result retrieval from completed persisted children with read_subagent_result. Each item may select a profile, model, effort, tools, write_paths, or read_only, and may declare depends_on to run after other items (research → implement → review). Tasks with no dependency between them run in parallel and must declare non-overlapping write_paths; ordered tasks may share paths. Omitted write_paths claim the whole workspace, so two or more concurrent writers without paths fail preflight before any task starts. A failed task's dependents are skipped; independent branches keep going unless fail_fast is set. Background mode returns a fleet job id collectable with wait.",
|
||||
"ReadOnly": false,
|
||||
"Schema": {
|
||||
"properties": {
|
||||
"fail_fast": {
|
||||
"description": "Stop starting new tasks after the first failure. Tasks already running are left to finish so partial writes are not abandoned mid-flight. Omitted (the default) means independent branches keep going; a failed task's dependents are skipped either way.",
|
||||
"type": "boolean"
|
||||
},
|
||||
"run_in_background": {
|
||||
"description": "Run the whole fleet asynchronously and return a job id collectable with wait. Items queue for concurrency/write slots inside the job.",
|
||||
"type": "boolean"
|
||||
@@ -435,6 +439,13 @@
|
||||
"description": "Array of 2–64 sub-tasks to run under the session scheduler.",
|
||||
"items": {
|
||||
"properties": {
|
||||
"depends_on": {
|
||||
"description": "Ids of tasks that must complete before this one starts. Unknown ids, self-edges, and cycles fail preflight. A task whose dependency fails or is skipped is skipped too. Ordered tasks may share write_paths; only tasks that can run at the same time need disjoint claims.",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
"type": "array"
|
||||
},
|
||||
"description": {
|
||||
"description": "Optional short label shown in the job list.",
|
||||
"type": "string"
|
||||
@@ -443,6 +454,10 @@
|
||||
"description": "Optional reasoning effort override.",
|
||||
"type": "string"
|
||||
},
|
||||
"id": {
|
||||
"description": "Optional stable id for this task, referenced by other tasks' depends_on. Defaults to the 1-based position.",
|
||||
"type": "string"
|
||||
},
|
||||
"max_steps": {
|
||||
"description": "Optional max tool-call rounds.",
|
||||
"minimum": 1,
|
||||
@@ -472,7 +487,7 @@
|
||||
"type": "array"
|
||||
},
|
||||
"write_paths": {
|
||||
"description": "Write targets for this item. Parallel writers must declare non-overlapping paths. Omitting write_paths claims the whole workspace; multiple whole-workspace claims (or any path overlap) fail preflight and start nothing.",
|
||||
"description": "Write targets for this item. Writers that can run at the same time must declare non-overlapping paths; writers ordered by depends_on may share them. Omitting write_paths claims the whole workspace; two concurrent whole-workspace claims (or any overlap between concurrent writers) fail preflight and start nothing.",
|
||||
"items": {
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user