From 0e62ab768451420a9e14f08fe735e8494dad801b Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 16:23:45 +0300 Subject: [PATCH] fix: persist all graph dispatch bindings before fan-out --- modules/work/application/graph_runtime.go | 21 ++++++++++++++------- tests/agent_execution/graph_runtime_test.go | 2 +- 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index 4517200..f50fde6 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -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 } diff --git a/tests/agent_execution/graph_runtime_test.go b/tests/agent_execution/graph_runtime_test.go index 5a07a8a..28f2880 100644 --- a/tests/agent_execution/graph_runtime_test.go +++ b/tests/agent_execution/graph_runtime_test.go @@ -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) } -- 2.54.0