feat(workflow): park parent when a WaitForOutput child yields no output

RunNode now honors NodeConfig.WaitForOutput: a child that opts in and
finishes without producing output parks the parent with ErrNodeInterrupted
instead of returning the zero value, so it re-runs and re-invokes RunNode
once the child can produce output. Mirrors adk-python's
ctx.run_node(raise_on_wait=True). Tracked via a sawOutput flag (nil is a
valid output) and a waitsForOutput helper reading the tri-state config.

Author: wolo-lab

workflow/dynamic_scheduler.go

@@ -186,6 +186,7 @@ func (s *dynamicSubScheduler) runNode(child Node, input any, opts runNodeOptions
 
 	var (
 		out         any
+		sawOutput   bool
 		interrupted bool
 	)
 	for ev, evErr := range child.Run(childCtx, input) {
@@ -216,6 +217,7 @@ func (s *dynamicSubScheduler) runNode(child Node, input any, opts runNodeOptions
 		}
 		if childOut, ok := childEventOutput(ev); ok {
 			out = childOut
+			sawOutput = true
 			// Stamp OutputFor so resume can attribute the output: the
 			// emitter's own path plus, under delegation, this parent and
 			// its ancestors (the parent then suppresses its own terminal
@@ -247,11 +249,29 @@ func (s *dynamicSubScheduler) runNode(child Node, input any, opts runNodeOptions
 		}
 	}
 
+	// A WaitForOutput child that produced no output is not done: park the
+	// parent (not terminal, not cached) so it re-runs and re-invokes
+	// RunNode once the child can produce output. Mirrors adk-python
+	// ctx.run_node(raise_on_wait=True).
+	if !sawOutput && waitsForOutput(child) {
+		return nil, &NodeRunError{
+			ChildName: name, ChildPath: childPath, RunID: runID,
+			Cause: ErrNodeInterrupted,
+		}
+	}
+
 	s.storeCachedOutput(childPath, out)
 	s.commitDelegation(childPath, out) // no-op unless this child claimed the delegation
 	return out, nil
 }
 
+// waitsForOutput reports whether node opts into WaitForOutput (tri-state
+// pointer; nil means the engine default of false).
+func waitsForOutput(node Node) bool {
+	w := node.Config().WaitForOutput
+	return w != nil && *w
+}
+
 func (s *dynamicSubScheduler) lookupCachedOutput(childPath string) (any, bool) {
 	s.mu.Lock()
 	defer s.mu.Unlock()

workflow/dynamic_scheduler_test.go

@@ -288,3 +288,41 @@ func (n *interruptThenFailNode) Run(agent.InvocationContext, any) iter.Seq2[*ses
 		yield(nil, errors.New("boom"))
 	}
 }
+
+// waitForOutputNode has WaitForOutput=true and emits a state-only event
+// (no Output), modeling an LlmAgent task/chat node that has not yet
+// produced its final output.
+type waitForOutputNode struct{ BaseNode }
+
+func newWaitForOutputNode(name string) *waitForOutputNode {
+	t := true
+	return &waitForOutputNode{BaseNode: NewBaseNode(name, "", NodeConfig{WaitForOutput: &t})}
+}
+
+func (n *waitForOutputNode) Run(agent.InvocationContext, any) iter.Seq2[*session.Event, error] {
+	return func(yield func(*session.Event, error) bool) {
+		yield(&session.Event{}, nil) // state-only: no Output, no RequestedInput
+	}
+}
+
+// waitForOutputWithValueNode has WaitForOutput=true and does emit an
+// Output, so RunNode must complete it normally.
+type waitForOutputWithValueNode struct {
+	BaseNode
+	out any
+}
+
+func newWaitForOutputWithValueNode(name string, out any) *waitForOutputWithValueNode {
+	t := true
+	return &waitForOutputWithValueNode{
+		BaseNode: NewBaseNode(name, "", NodeConfig{WaitForOutput: &t}),
+		out:      out,
+	}
+}
+
+func (n *waitForOutputWithValueNode) Run(agent.InvocationContext, any) iter.Seq2[*session.Event, error] {
+	out := n.out
+	return func(yield func(*session.Event, error) bool) {
+		yield(&session.Event{Output: out}, nil)
+	}
+}

workflow/run_node_test.go

@@ -67,6 +67,46 @@ func TestRunNode_PropagatesErrNodeInterrupted(t *testing.T) {
 	}
 }
 
+func TestRunNode_WaitForOutputChildWithNoOutput_ParksParent(t *testing.T) {
+	// A WaitForOutput child that finishes without producing output must
+	// park the parent (ErrNodeInterrupted), not falsely complete it with
+	// the zero value. Mirrors adk-python ctx.run_node(raise_on_wait=True).
+	child := newWaitForOutputNode("waiter")
+	_, err := runInOrchestratorWithErr[string](t, func(ctx NodeContext) (string, error) {
+		return RunNode[string](ctx, child, nil)
+	})
+	if !errors.Is(err, ErrNodeInterrupted) {
+		t.Errorf("err = %v, want errors.Is ErrNodeInterrupted", err)
+	}
+}
+
+func TestRunNode_WaitForOutputChildWithOutput_Completes(t *testing.T) {
+	// A WaitForOutput child that does produce output completes normally;
+	// the gate must only fire on missing output.
+	child := newWaitForOutputWithValueNode("waiter", "done")
+	got := runInOrchestrator[string](t, func(ctx NodeContext) (string, error) {
+		return RunNode[string](ctx, child, nil)
+	})
+	if got != "done" {
+		t.Errorf("RunNode output = %q, want %q", got, "done")
+	}
+}
+
+func TestRunNode_NoWaitForOutputChildWithNoOutput_ReturnsZero(t *testing.T) {
+	// Without WaitForOutput, a child that emits no output still completes
+	// and yields the zero value — the gate must not change this default.
+	child := newStubNode("c", nil)
+	got, err := runInOrchestratorWithErr[string](t, func(ctx NodeContext) (string, error) {
+		return RunNode[string](ctx, child, nil)
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if got != "" {
+		t.Errorf("RunNode output = %q, want zero value", got)
+	}
+}
+
 func TestRunNode_PropagatesErrNodeFailed(t *testing.T) {
 	failer := newFailingNode("failer", errors.New("boom"))
 	_, err := runInOrchestratorWithErr[string](t, func(ctx NodeContext) (string, error) {