m/Assistant
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: persist all graph dispatch bindings before fan-out #119

Merged
m merged 2 commits from codex/wp-400-graph-runtime into main 2026-08-22 13:25:24 +00:00
Conversation 0 Commits 2 Files Changed 2
+15 -8
2 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/work/application/graph_runtime.go
+14 -7
View File
@@ -44,6 +44,19 @@ func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request Graph
for _, intent := range request.Intents {
intentByDispatch[intent.DispatchID] = intent
}
for _, node := range admitted.Nodes {
if node.DispatchID == "" {
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
}
if _, ok := intentByDispatch[node.DispatchID]; !ok {
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
}
}
for _, intent := range request.Intents {
if err := store.SaveDispatchIntent(ctx, intent); err != nil {
return work.WorkGraph{}, nil, err
}
}
next := admitted
next.Revision++
next.State = work.GraphRunning
@@ -58,13 +71,7 @@ func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request Graph
if next.Nodes[i].DispatchID == "" {
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
}
intent, ok := intentByDispatch[next.Nodes[i].DispatchID]
if !ok {
return work.WorkGraph{}, nil, ErrGraphDispatchBinding
}
if err := store.SaveDispatchIntent(ctx, intent); err != nil {
return work.WorkGraph{}, nil, err
}
intent := intentByDispatch[next.Nodes[i].DispatchID]
if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: intent.DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: request.Now, UpdatedAtUnixMillis: request.Now}); err != nil {
return work.WorkGraph{}, nil, err
}
tests/agent_execution/graph_runtime_test.go
+1 -1
View File
@@ -40,7 +40,7 @@ func itoa(value int64) string {
func TestGraphRuntimeSchedulesFanOutAndCompletesFanIn(t *testing.T) {
store := &graphStore{}
graph, ready, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "g", Revision: 1, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "a", DispatchID: "d-a", IdempotencyKey: "a", State: work.GraphNodeReady}, {ID: "b", DispatchID: "d-b", IdempotencyKey: "b", State: work.GraphNodeReady}, {ID: "join", DispatchID: "d-join", IdempotencyKey: "join", State: work.GraphNodeBlocked}}, Edges: []work.GraphEdge{{From: "a", To: "join"}, {From: "b", To: "join"}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "d-a"}, {DispatchID: "d-b"}}, Now: 1})
graph, ready, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "g", Revision: 1, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "a", DispatchID: "d-a", IdempotencyKey: "a", State: work.GraphNodeReady}, {ID: "b", DispatchID: "d-b", IdempotencyKey: "b", State: work.GraphNodeReady}, {ID: "join", DispatchID: "d-join", IdempotencyKey: "join", State: work.GraphNodeBlocked}}, Edges: []work.GraphEdge{{From: "a", To: "join"}, {From: "b", To: "join"}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "d-a"}, {DispatchID: "d-b"}, {DispatchID: "d-join"}}, Now: 1})
if err != nil || len(ready) != 2 || len(store.queued) != 2 {
t.Fatalf("graph=%+v ready=%v queued=%v err=%v", graph, ready, store.queued, err)
}