diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index 77827b4..4517200 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -106,6 +106,12 @@ func CompleteGraphNode(ctx context.Context, store ports.GraphStore, graphID stri for _, candidate := range ready { if graph.Nodes[i].ID == candidate && graph.Nodes[i].State == work.GraphNodeBlocked { graph.Nodes[i].State = work.GraphNodeReady + if graph.Nodes[i].DispatchID == "" { + return work.WorkGraph{}, ErrGraphDispatchBinding + } + if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: graph.Nodes[i].DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: time.Now().UnixMilli(), UpdatedAtUnixMillis: time.Now().UnixMilli()}); err != nil { + return work.WorkGraph{}, err + } } } } diff --git a/tests/agent_execution/graph_runtime_test.go b/tests/agent_execution/graph_runtime_test.go index c7c99ea..5a07a8a 100644 --- a/tests/agent_execution/graph_runtime_test.go +++ b/tests/agent_execution/graph_runtime_test.go @@ -50,6 +50,9 @@ func TestGraphRuntimeSchedulesFanOutAndCompletesFanIn(t *testing.T) { if _, err := application.CompleteGraphNode(context.Background(), store, "g", 3, "b", true); err != nil { t.Fatal(err) } + if len(store.queued) != 3 || store.queued[2] != "d-join" { + t.Fatalf("fan-in dispatch was not enqueued: %v", store.queued) + } completed, err := application.CompleteGraphNode(context.Background(), store, "g", 4, "join", true) if err != nil || completed.State != work.GraphCompleted { t.Fatalf("completed=%+v err=%v", completed, err)