From 073cc6a826dbe13c39e6cf7ab689ad9b37b05f11 Mon Sep 17 00:00:00 2001 From: lonestarx1 Date: Mon, 16 Feb 2026 19:10:43 +0900 Subject: [PATCH 1/2] Implement Phase 6: dynamic orchestration with resource governance Add the dynamic orchestration pattern enabling agents to spawn child agents, teams, pipelines, or graphs at runtime. The Runtime manages resource governance: concurrency limits (semaphore), maximum nesting depth, and cost budgets across all spawned children. Includes async spawning via Go/Future, context-based runtime propagation, cascading cancellation, and aggregate cost/usage metrics. 27 tests covering all spawn types, depth enforcement, cost budgets, concurrency limits, async futures, cancellation, and tracing. Updates website docs with Dynamic Orchestration, Resource Governance, and Async & Futures sections. Adds dynamic research coordinator example. --- README.md | 2 +- pkg/orchestrator/dynamic/doc.go | 38 ++ pkg/orchestrator/dynamic/runtime.go | 419 +++++++++++++ pkg/orchestrator/dynamic/runtime_test.go | 763 +++++++++++++++++++++++ website/app/docs/page.tsx | 181 +++++- website/app/examples/page.tsx | 135 ++++ website/components/Architecture.tsx | 2 +- 7 files changed, 1537 insertions(+), 3 deletions(-) create mode 100644 pkg/orchestrator/dynamic/doc.go create mode 100644 pkg/orchestrator/dynamic/runtime.go create mode 100644 pkg/orchestrator/dynamic/runtime_test.go diff --git a/README.md b/README.md index e394d8e..6ae8f75 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ Sequential handoff between specialists. Each agent completes its work, yields it Like a pipeline, but with conditional edges, parallel fan-out, fan-in merging, and loops. Agents execute concurrently in waves — when a node completes, its outgoing edges are evaluated and successor nodes fire when all dependencies are satisfied. Supports configurable iteration limits, cost budgets, timeouts, and exports to Graphviz DOT format for visualization. ### Dynamic Orchestration -GoGrid's most powerful pattern. Agents can spawn child agents, child teams, child pipelines, or child graphs dynamically at runtime. Unlimited scaling with minimal assumptions about how a problem gets solved. For when the developer doesn't know — or shouldn't hardcode — the exact steps to a solution. +GoGrid's most powerful pattern. A Runtime enables agents to spawn child agents, teams, pipelines, or graphs dynamically at runtime. Resource governance controls concurrency limits, nesting depth, and cost budgets across all spawned children. Async futures allow parallel child execution with aggregate metrics tracking. > All GoGrid patterns are composable. A team can contain pipelines. A graph node can spawn a dynamic orchestrator. The architecture adapts to the problem, not the other way around. diff --git a/pkg/orchestrator/dynamic/doc.go b/pkg/orchestrator/dynamic/doc.go new file mode 100644 index 0000000..6303a5a --- /dev/null +++ b/pkg/orchestrator/dynamic/doc.go @@ -0,0 +1,38 @@ +// Package dynamic implements GoGrid's Dynamic Orchestration pattern. +// +// Dynamic orchestration enables agents to spawn child agents, teams, +// pipelines, or graphs at runtime. This is GoGrid's most powerful +// pattern — the executing agent decides which orchestration to use +// based on the problem at hand. +// +// A Runtime manages resource governance: concurrency limits, maximum +// nesting depth, cost budgets, and cascading cancellation. Child +// orchestrations inherit the parent's tracing context and are tracked +// for aggregate cost and usage metrics. +// +// Usage: +// +// rt := dynamic.New("coordinator", +// dynamic.WithConfig(dynamic.Config{ +// MaxConcurrent: 5, +// MaxDepth: 3, +// CostBudget: 1.00, +// }), +// ) +// ctx := rt.Context(ctx) +// result, err := rt.SpawnAgent(ctx, researchAgent, "Find papers on X") +// +// For async spawning, use Go to launch children in the background: +// +// f := rt.Go(ctx, "research", func(ctx context.Context) (string, error) { +// r, err := rt.SpawnAgent(ctx, researchAgent, input) +// if err != nil { +// return "", err +// } +// return r.Message.Content, nil +// }) +// output, err := f.Wait(ctx) +// +// The Runtime is made available to child orchestrations via context, +// enabling nested dynamic spawning up to the configured MaxDepth. +package dynamic diff --git a/pkg/orchestrator/dynamic/runtime.go b/pkg/orchestrator/dynamic/runtime.go new file mode 100644 index 0000000..21e968d --- /dev/null +++ b/pkg/orchestrator/dynamic/runtime.go @@ -0,0 +1,419 @@ +package dynamic + +import ( + "context" + "errors" + "fmt" + "strconv" + "sync" + + "github.com/lonestarx1/gogrid/internal/id" + "github.com/lonestarx1/gogrid/pkg/agent" + "github.com/lonestarx1/gogrid/pkg/llm" + "github.com/lonestarx1/gogrid/pkg/orchestrator/graph" + "github.com/lonestarx1/gogrid/pkg/orchestrator/pipeline" + "github.com/lonestarx1/gogrid/pkg/orchestrator/team" + "github.com/lonestarx1/gogrid/pkg/trace" +) + +// Sentinel errors for resource governance violations. +var ( + // ErrMaxDepth is returned when a spawn would exceed the maximum + // nesting depth. + ErrMaxDepth = errors.New("dynamic: maximum nesting depth exceeded") + // ErrCostBudget is returned when the runtime's cost budget is + // exhausted and no more children can be spawned. + ErrCostBudget = errors.New("dynamic: cost budget exhausted") +) + +type runtimeKey struct{} +type depthKey struct{} + +// FromContext retrieves the Runtime from the context, or nil if none. +func FromContext(ctx context.Context) *Runtime { + r, _ := ctx.Value(runtimeKey{}).(*Runtime) + return r +} + +// DepthFromContext returns the current nesting depth from the context. +// Returns 0 if no depth has been set. +func DepthFromContext(ctx context.Context) int { + d, _ := ctx.Value(depthKey{}).(int) + return d +} + +// Config controls resource governance for dynamic orchestration. +type Config struct { + // MaxConcurrent is the maximum number of children that can execute + // simultaneously. 0 means no limit. + MaxConcurrent int + // MaxDepth is the maximum nesting depth for recursive spawning. + // 0 defaults to 10. + MaxDepth int + // CostBudget is the maximum total cost in USD across all children. + // 0 means no limit. + CostBudget float64 +} + +// ChildResult records the outcome of a single spawned child. +type ChildResult struct { + // Name identifies the child. + Name string + // Type is "agent", "team", "pipeline", or "graph". + Type string + // Output is the child's final output content. + Output string + // Cost is the child's total cost in USD. + Cost float64 + // Usage is the child's token usage. + Usage llm.Usage + // Error is non-nil if the child failed. + Error error +} + +// Result holds aggregate metrics from all children spawned by a Runtime. +type Result struct { + // RunID uniquely identifies this runtime execution. + RunID string + // Children lists all spawned child results in order. + Children []ChildResult + // TotalCost is the aggregate cost across all children. + TotalCost float64 + // TotalUsage is the aggregate token usage across all children. + TotalUsage llm.Usage +} + +// Option is a functional option for configuring a Runtime. +type Option func(*Runtime) + +// WithConfig sets the runtime's resource governance configuration. +func WithConfig(c Config) Option { + return func(r *Runtime) { + r.config = c + } +} + +// WithTracer sets the tracer for observability. +func WithTracer(t trace.Tracer) Option { + return func(r *Runtime) { + r.tracer = t + } +} + +// Runtime enables dynamic spawning of child orchestrations with +// resource governance. It tracks concurrency, nesting depth, cost +// budgets, and aggregate metrics across all spawned children. +type Runtime struct { + name string + tracer trace.Tracer + config Config + runID string + + mu sync.Mutex + children []ChildResult + totalCost float64 + totalUsage llm.Usage + + sem chan struct{} // concurrency semaphore, nil if unlimited + wg sync.WaitGroup +} + +// New creates a Runtime with the given name and options. +func New(name string, opts ...Option) *Runtime { + r := &Runtime{ + name: name, + tracer: trace.Noop{}, + runID: id.New(), + } + for _, opt := range opts { + opt(r) + } + if r.config.MaxConcurrent > 0 { + r.sem = make(chan struct{}, r.config.MaxConcurrent) + } + if r.config.MaxDepth <= 0 { + r.config.MaxDepth = 10 + } + return r +} + +// Name returns the runtime's name. +func (r *Runtime) Name() string { return r.name } + +// Context returns a new context with this runtime embedded. +// Child orchestrations can retrieve it with FromContext. +func (r *Runtime) Context(ctx context.Context) context.Context { + return context.WithValue(ctx, runtimeKey{}, r) +} + +// RemainingBudget returns the remaining cost budget in USD. +// Returns -1 if no budget limit is configured. +func (r *Runtime) RemainingBudget() float64 { + if r.config.CostBudget <= 0 { + return -1 + } + r.mu.Lock() + defer r.mu.Unlock() + remaining := r.config.CostBudget - r.totalCost + if remaining < 0 { + return 0 + } + return remaining +} + +// SpawnAgent runs a single agent as a child of this runtime. +func (r *Runtime) SpawnAgent(ctx context.Context, a *agent.Agent, input string) (*agent.Result, error) { + if err := r.checkLimits(ctx); err != nil { + return nil, err + } + if err := r.acquireSlot(ctx); err != nil { + return nil, err + } + defer r.releaseSlot() + + ctx, span := r.tracer.StartSpan(ctx, "dynamic.spawn_agent") + span.SetAttribute("dynamic.child.name", a.Name()) + span.SetAttribute("dynamic.child.type", "agent") + span.SetAttribute("dynamic.depth", strconv.Itoa(DepthFromContext(ctx))) + defer r.tracer.EndSpan(span) + + childCtx := r.childContext(ctx) + result, err := a.Run(childCtx, input) + if err != nil { + span.SetError(err) + r.recordChild(a.Name(), "agent", "", 0, llm.Usage{}, err) + return nil, fmt.Errorf("dynamic spawn agent %q: %w", a.Name(), err) + } + + r.recordChild(a.Name(), "agent", result.Message.Content, result.Cost, result.Usage, nil) + span.SetAttribute("dynamic.child.cost_usd", fmt.Sprintf("%.6f", result.Cost)) + + return result, nil +} + +// SpawnTeam runs a team as a child of this runtime. +func (r *Runtime) SpawnTeam(ctx context.Context, t *team.Team, input string) (*team.Result, error) { + if err := r.checkLimits(ctx); err != nil { + return nil, err + } + if err := r.acquireSlot(ctx); err != nil { + return nil, err + } + defer r.releaseSlot() + + ctx, span := r.tracer.StartSpan(ctx, "dynamic.spawn_team") + span.SetAttribute("dynamic.child.name", t.Name()) + span.SetAttribute("dynamic.child.type", "team") + span.SetAttribute("dynamic.depth", strconv.Itoa(DepthFromContext(ctx))) + defer r.tracer.EndSpan(span) + + childCtx := r.childContext(ctx) + result, err := t.Run(childCtx, input) + if err != nil { + span.SetError(err) + r.recordChild(t.Name(), "team", "", 0, llm.Usage{}, err) + return nil, fmt.Errorf("dynamic spawn team %q: %w", t.Name(), err) + } + + r.recordChild(t.Name(), "team", result.Decision.Content, result.TotalCost, result.TotalUsage, nil) + span.SetAttribute("dynamic.child.cost_usd", fmt.Sprintf("%.6f", result.TotalCost)) + + return result, nil +} + +// SpawnPipeline runs a pipeline as a child of this runtime. +func (r *Runtime) SpawnPipeline(ctx context.Context, p *pipeline.Pipeline, input string) (*pipeline.Result, error) { + if err := r.checkLimits(ctx); err != nil { + return nil, err + } + if err := r.acquireSlot(ctx); err != nil { + return nil, err + } + defer r.releaseSlot() + + ctx, span := r.tracer.StartSpan(ctx, "dynamic.spawn_pipeline") + span.SetAttribute("dynamic.child.name", p.Name()) + span.SetAttribute("dynamic.child.type", "pipeline") + span.SetAttribute("dynamic.depth", strconv.Itoa(DepthFromContext(ctx))) + defer r.tracer.EndSpan(span) + + childCtx := r.childContext(ctx) + result, err := p.Run(childCtx, input) + if err != nil { + span.SetError(err) + r.recordChild(p.Name(), "pipeline", "", 0, llm.Usage{}, err) + return nil, fmt.Errorf("dynamic spawn pipeline %q: %w", p.Name(), err) + } + + r.recordChild(p.Name(), "pipeline", result.Output, result.TotalCost, result.TotalUsage, nil) + span.SetAttribute("dynamic.child.cost_usd", fmt.Sprintf("%.6f", result.TotalCost)) + + return result, nil +} + +// SpawnGraph runs a graph as a child of this runtime. +func (r *Runtime) SpawnGraph(ctx context.Context, g *graph.Graph, input string) (*graph.Result, error) { + if err := r.checkLimits(ctx); err != nil { + return nil, err + } + if err := r.acquireSlot(ctx); err != nil { + return nil, err + } + defer r.releaseSlot() + + ctx, span := r.tracer.StartSpan(ctx, "dynamic.spawn_graph") + span.SetAttribute("dynamic.child.name", g.Name()) + span.SetAttribute("dynamic.child.type", "graph") + span.SetAttribute("dynamic.depth", strconv.Itoa(DepthFromContext(ctx))) + defer r.tracer.EndSpan(span) + + childCtx := r.childContext(ctx) + result, err := g.Run(childCtx, input) + if err != nil { + span.SetError(err) + r.recordChild(g.Name(), "graph", "", 0, llm.Usage{}, err) + return nil, fmt.Errorf("dynamic spawn graph %q: %w", g.Name(), err) + } + + r.recordChild(g.Name(), "graph", result.Output, result.TotalCost, result.TotalUsage, nil) + span.SetAttribute("dynamic.child.cost_usd", fmt.Sprintf("%.6f", result.TotalCost)) + + return result, nil +} + +// Future represents an asynchronous child operation launched via Go. +type Future struct { + ch chan struct{} + once sync.Once + output string + err error +} + +func newFuture() *Future { + return &Future{ch: make(chan struct{})} +} + +func (f *Future) complete(output string, err error) { + f.once.Do(func() { + f.output = output + f.err = err + close(f.ch) + }) +} + +// Wait blocks until the future completes or the context is canceled. +// Returns the output string and any error from the spawned function. +func (f *Future) Wait(ctx context.Context) (string, error) { + select { + case <-f.ch: + return f.output, f.err + case <-ctx.Done(): + return "", ctx.Err() + } +} + +// Done returns a channel that is closed when the future completes. +func (f *Future) Done() <-chan struct{} { + return f.ch +} + +// Go launches fn as a background child of this runtime. +// The function runs in a new goroutine. Use the returned Future to +// await the result, or call Wait to block until all background +// children complete. +func (r *Runtime) Go(ctx context.Context, name string, fn func(ctx context.Context) (string, error)) *Future { + f := newFuture() + _, span := r.tracer.StartSpan(ctx, "dynamic.go") + span.SetAttribute("dynamic.child.name", name) + + r.wg.Add(1) + go func() { + defer r.wg.Done() + defer r.tracer.EndSpan(span) + + output, err := fn(ctx) + if err != nil { + span.SetError(err) + } + f.complete(output, err) + }() + + return f +} + +// Wait blocks until all background children launched via Go complete. +func (r *Runtime) Wait() { + r.wg.Wait() +} + +// Result returns aggregate metrics from all spawned children. +func (r *Runtime) Result() *Result { + r.mu.Lock() + defer r.mu.Unlock() + children := make([]ChildResult, len(r.children)) + copy(children, r.children) + return &Result{ + RunID: r.runID, + Children: children, + TotalCost: r.totalCost, + TotalUsage: r.totalUsage, + } +} + +// checkLimits verifies that depth and budget constraints are met. +func (r *Runtime) checkLimits(ctx context.Context) error { + depth := DepthFromContext(ctx) + if depth >= r.config.MaxDepth { + return ErrMaxDepth + } + r.mu.Lock() + defer r.mu.Unlock() + if r.config.CostBudget > 0 && r.totalCost >= r.config.CostBudget { + return ErrCostBudget + } + return nil +} + +// acquireSlot acquires a concurrency slot, blocking until one is +// available or the context is canceled. +func (r *Runtime) acquireSlot(ctx context.Context) error { + if r.sem == nil { + return nil + } + select { + case r.sem <- struct{}{}: + return nil + case <-ctx.Done(): + return fmt.Errorf("dynamic: waiting for concurrency slot: %w", ctx.Err()) + } +} + +func (r *Runtime) releaseSlot() { + if r.sem != nil { + <-r.sem + } +} + +// childContext creates a child context with incremented depth. +func (r *Runtime) childContext(ctx context.Context) context.Context { + depth := DepthFromContext(ctx) + return context.WithValue(ctx, depthKey{}, depth+1) +} + +// recordChild records a child execution result and updates aggregates. +func (r *Runtime) recordChild(name, typ, output string, cost float64, usage llm.Usage, err error) { + r.mu.Lock() + defer r.mu.Unlock() + r.children = append(r.children, ChildResult{ + Name: name, + Type: typ, + Output: output, + Cost: cost, + Usage: usage, + Error: err, + }) + r.totalCost += cost + r.totalUsage.PromptTokens += usage.PromptTokens + r.totalUsage.CompletionTokens += usage.CompletionTokens + r.totalUsage.TotalTokens += usage.TotalTokens +} diff --git a/pkg/orchestrator/dynamic/runtime_test.go b/pkg/orchestrator/dynamic/runtime_test.go new file mode 100644 index 0000000..8537989 --- /dev/null +++ b/pkg/orchestrator/dynamic/runtime_test.go @@ -0,0 +1,763 @@ +package dynamic + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/lonestarx1/gogrid/pkg/agent" + "github.com/lonestarx1/gogrid/pkg/llm" + "github.com/lonestarx1/gogrid/pkg/orchestrator/graph" + "github.com/lonestarx1/gogrid/pkg/orchestrator/pipeline" + "github.com/lonestarx1/gogrid/pkg/orchestrator/team" + "github.com/lonestarx1/gogrid/pkg/trace" +) + +// --- Mock providers --- + +type mockProvider struct { + response *llm.Response + calls atomic.Int32 +} + +func newMockProvider(content string) *mockProvider { + return &mockProvider{ + response: &llm.Response{ + Message: llm.NewAssistantMessage(content), + Usage: llm.Usage{PromptTokens: 10, CompletionTokens: 5, TotalTokens: 15}, + Model: "mock-model", + }, + } +} + +func (m *mockProvider) Complete(_ context.Context, _ llm.Params) (*llm.Response, error) { + m.calls.Add(1) + return m.response, nil +} + +type slowProvider struct { + delay time.Duration + response *llm.Response +} + +func (s *slowProvider) Complete(ctx context.Context, _ llm.Params) (*llm.Response, error) { + select { + case <-time.After(s.delay): + return s.response, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +type errorProvider struct{ err error } + +func (e *errorProvider) Complete(_ context.Context, _ llm.Params) (*llm.Response, error) { + return nil, e.err +} + +// costProvider returns a response with configurable cost. +type costProvider struct { + content string + usage llm.Usage +} + +func (c *costProvider) Complete(_ context.Context, _ llm.Params) (*llm.Response, error) { + return &llm.Response{ + Message: llm.NewAssistantMessage(c.content), + Usage: c.usage, + Model: "mock-model", + }, nil +} + +// --- Helpers --- + +func newTestAgent(name, content string) *agent.Agent { + return agent.New(name, + agent.WithProvider(newMockProvider(content)), + agent.WithModel("mock-model"), + ) +} + +func newSlowAgent(name string, delay time.Duration) *agent.Agent { + return agent.New(name, + agent.WithProvider(&slowProvider{ + delay: delay, + response: &llm.Response{ + Message: llm.NewAssistantMessage(name + "-done"), + Usage: llm.Usage{PromptTokens: 10, CompletionTokens: 5, TotalTokens: 15}, + Model: "mock-model", + }, + }), + agent.WithModel("mock-model"), + ) +} + +func newTestTeam(name string, agentContent string) *team.Team { + return team.New(name, + team.WithMembers( + team.Member{Agent: newTestAgent("member-a", agentContent)}, + team.Member{Agent: newTestAgent("member-b", agentContent)}, + ), + ) +} + +func newTestPipeline(name string) *pipeline.Pipeline { + return pipeline.New(name, + pipeline.WithStages( + pipeline.Stage{ + Name: "stage-1", + Agent: newTestAgent("s1", "stage-1-output"), + }, + pipeline.Stage{ + Name: "stage-2", + Agent: newTestAgent("s2", "stage-2-output"), + }, + ), + ) +} + +func newTestGraph(name string) *graph.Graph { + g, _ := graph.NewBuilder(name). + AddNode("draft", newTestAgent("draft", "draft-output")). + AddNode("review", newTestAgent("review", "review-output")). + AddEdge("draft", "review"). + Build() + return g +} + +// --- Tests --- + +func TestNew(t *testing.T) { + rt := New("test-runtime") + if rt.Name() != "test-runtime" { + t.Errorf("Name = %q, want %q", rt.Name(), "test-runtime") + } + if rt.config.MaxDepth != 10 { + t.Errorf("MaxDepth = %d, want default 10", rt.config.MaxDepth) + } + if rt.sem != nil { + t.Error("sem should be nil when MaxConcurrent is 0") + } +} + +func TestNewWithConfig(t *testing.T) { + rt := New("rt", + WithConfig(Config{ + MaxConcurrent: 3, + MaxDepth: 5, + CostBudget: 1.50, + }), + ) + if rt.config.MaxConcurrent != 3 { + t.Errorf("MaxConcurrent = %d, want 3", rt.config.MaxConcurrent) + } + if rt.config.MaxDepth != 5 { + t.Errorf("MaxDepth = %d, want 5", rt.config.MaxDepth) + } + if rt.config.CostBudget != 1.50 { + t.Errorf("CostBudget = %f, want 1.50", rt.config.CostBudget) + } + if rt.sem == nil || cap(rt.sem) != 3 { + t.Error("sem should have capacity 3") + } +} + +func TestContext(t *testing.T) { + rt := New("rt") + ctx := rt.Context(context.Background()) + + got := FromContext(ctx) + if got != rt { + t.Error("FromContext returned different runtime") + } + + // No runtime in plain context. + if FromContext(context.Background()) != nil { + t.Error("FromContext should return nil for plain context") + } +} + +func TestDepthFromContext(t *testing.T) { + ctx := context.Background() + if DepthFromContext(ctx) != 0 { + t.Error("DepthFromContext should return 0 for plain context") + } + + ctx = context.WithValue(ctx, depthKey{}, 3) + if DepthFromContext(ctx) != 3 { + t.Errorf("DepthFromContext = %d, want 3", DepthFromContext(ctx)) + } +} + +func TestSpawnAgent(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + a := newTestAgent("researcher", "research-result") + result, err := rt.SpawnAgent(ctx, a, "find papers") + if err != nil { + t.Fatalf("SpawnAgent error: %v", err) + } + if result.Message.Content != "research-result" { + t.Errorf("output = %q, want %q", result.Message.Content, "research-result") + } + + // Check child recorded. + res := rt.Result() + if len(res.Children) != 1 { + t.Fatalf("Children = %d, want 1", len(res.Children)) + } + if res.Children[0].Name != "researcher" { + t.Errorf("child name = %q, want %q", res.Children[0].Name, "researcher") + } + if res.Children[0].Type != "agent" { + t.Errorf("child type = %q, want %q", res.Children[0].Type, "agent") + } + if res.Children[0].Output != "research-result" { + t.Errorf("child output = %q, want %q", res.Children[0].Output, "research-result") + } +} + +func TestSpawnAgentError(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + a := agent.New("broken", + agent.WithProvider(&errorProvider{err: errors.New("llm down")}), + agent.WithModel("mock-model"), + ) + + _, err := rt.SpawnAgent(ctx, a, "input") + if err == nil { + t.Fatal("expected error from SpawnAgent") + } + + // Error child should be recorded. + res := rt.Result() + if len(res.Children) != 1 { + t.Fatalf("Children = %d, want 1", len(res.Children)) + } + if res.Children[0].Error == nil { + t.Error("child error should be non-nil") + } +} + +func TestSpawnTeam(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + tm := newTestTeam("debate", "team-response") + result, err := rt.SpawnTeam(ctx, tm, "discuss topic") + if err != nil { + t.Fatalf("SpawnTeam error: %v", err) + } + if result.Decision.Content == "" { + t.Error("team decision should not be empty") + } + + res := rt.Result() + if len(res.Children) != 1 { + t.Fatalf("Children = %d, want 1", len(res.Children)) + } + if res.Children[0].Type != "team" { + t.Errorf("child type = %q, want %q", res.Children[0].Type, "team") + } +} + +func TestSpawnPipeline(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + p := newTestPipeline("research-pipeline") + result, err := rt.SpawnPipeline(ctx, p, "research input") + if err != nil { + t.Fatalf("SpawnPipeline error: %v", err) + } + if result.Output != "stage-2-output" { + t.Errorf("output = %q, want %q", result.Output, "stage-2-output") + } + + res := rt.Result() + if len(res.Children) != 1 { + t.Fatalf("Children = %d, want 1", len(res.Children)) + } + if res.Children[0].Type != "pipeline" { + t.Errorf("child type = %q, want %q", res.Children[0].Type, "pipeline") + } +} + +func TestSpawnGraph(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + g := newTestGraph("review-graph") + result, err := rt.SpawnGraph(ctx, g, "write something") + if err != nil { + t.Fatalf("SpawnGraph error: %v", err) + } + if result.Output != "review-output" { + t.Errorf("output = %q, want %q", result.Output, "review-output") + } + + res := rt.Result() + if len(res.Children) != 1 { + t.Fatalf("Children = %d, want 1", len(res.Children)) + } + if res.Children[0].Type != "graph" { + t.Errorf("child type = %q, want %q", res.Children[0].Type, "graph") + } +} + +func TestMaxDepthEnforcement(t *testing.T) { + rt := New("rt", WithConfig(Config{MaxDepth: 2})) + + // Depth 0 → OK. + ctx := context.Background() + _, err := rt.SpawnAgent(ctx, newTestAgent("a1", "ok"), "input") + if err != nil { + t.Fatalf("depth 0 should succeed: %v", err) + } + + // Depth 1 → OK. + ctx1 := context.WithValue(ctx, depthKey{}, 1) + _, err = rt.SpawnAgent(ctx1, newTestAgent("a2", "ok"), "input") + if err != nil { + t.Fatalf("depth 1 should succeed: %v", err) + } + + // Depth 2 → exceeds MaxDepth=2. + ctx2 := context.WithValue(ctx, depthKey{}, 2) + _, err = rt.SpawnAgent(ctx2, newTestAgent("a3", "fail"), "input") + if !errors.Is(err, ErrMaxDepth) { + t.Errorf("depth 2 error = %v, want ErrMaxDepth", err) + } +} + +func TestCostBudgetEnforcement(t *testing.T) { + rt := New("rt", WithConfig(Config{CostBudget: 0.001})) + ctx := context.Background() + + // First spawn succeeds (cost is 0 before). + _, err := rt.SpawnAgent(ctx, newTestAgent("a1", "ok"), "input") + if err != nil { + t.Fatalf("first spawn should succeed: %v", err) + } + + // The mock agent has some cost from the cost tracker. + // Second spawn should fail because budget is tiny and first spawn used some. + // Force the cost to exceed budget. + rt.mu.Lock() + rt.totalCost = 0.001 // Force to budget limit. + rt.mu.Unlock() + + _, err = rt.SpawnAgent(ctx, newTestAgent("a2", "fail"), "input") + if !errors.Is(err, ErrCostBudget) { + t.Errorf("second spawn error = %v, want ErrCostBudget", err) + } +} + +func TestMaxConcurrentEnforcement(t *testing.T) { + rt := New("rt", WithConfig(Config{MaxConcurrent: 2})) + ctx := context.Background() + + // Fill both slots. + slow1 := newSlowAgent("slow1", 200*time.Millisecond) + slow2 := newSlowAgent("slow2", 200*time.Millisecond) + + done := make(chan struct{}) + go func() { + _, _ = rt.SpawnAgent(ctx, slow1, "input") + done <- struct{}{} + }() + go func() { + _, _ = rt.SpawnAgent(ctx, slow2, "input") + done <- struct{}{} + }() + + // Give the goroutines time to start and acquire slots. + time.Sleep(50 * time.Millisecond) + + // Third spawn with a short timeout should fail because slots are full. + shortCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + _, err := rt.SpawnAgent(shortCtx, newTestAgent("blocked", "fail"), "input") + if err == nil { + t.Error("expected error when all slots are full") + } + + // Wait for slow agents to finish. + <-done + <-done +} + +func TestCascadingCancellation(t *testing.T) { + rt := New("rt") + ctx, cancel := context.WithCancel(context.Background()) + + slow := newSlowAgent("slow", 5*time.Second) + + errCh := make(chan error, 1) + go func() { + _, err := rt.SpawnAgent(ctx, slow, "input") + errCh <- err + }() + + // Cancel the parent context. + time.Sleep(50 * time.Millisecond) + cancel() + + err := <-errCh + if err == nil { + t.Fatal("expected error from canceled context") + } +} + +func TestGoAndFuture(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + f := rt.Go(ctx, "async-task", func(ctx context.Context) (string, error) { + return "async-result", nil + }) + + output, err := f.Wait(ctx) + if err != nil { + t.Fatalf("Future.Wait error: %v", err) + } + if output != "async-result" { + t.Errorf("output = %q, want %q", output, "async-result") + } +} + +func TestGoWithSpawn(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + f := rt.Go(ctx, "spawn-async", func(ctx context.Context) (string, error) { + result, err := rt.SpawnAgent(ctx, newTestAgent("inner", "inner-result"), "input") + if err != nil { + return "", err + } + return result.Message.Content, nil + }) + + output, err := f.Wait(ctx) + if err != nil { + t.Fatalf("Future.Wait error: %v", err) + } + if output != "inner-result" { + t.Errorf("output = %q, want %q", output, "inner-result") + } + + // The SpawnAgent call should have recorded a child. + res := rt.Result() + if len(res.Children) != 1 { + t.Fatalf("Children = %d, want 1", len(res.Children)) + } +} + +func TestGoError(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + f := rt.Go(ctx, "fail-task", func(ctx context.Context) (string, error) { + return "", errors.New("task failed") + }) + + _, err := f.Wait(ctx) + if err == nil || err.Error() != "task failed" { + t.Errorf("error = %v, want 'task failed'", err) + } +} + +func TestFutureContextCancel(t *testing.T) { + rt := New("rt") + goCtx, goCancel := context.WithCancel(context.Background()) + + f := rt.Go(goCtx, "slow-task", func(ctx context.Context) (string, error) { + select { + case <-time.After(10 * time.Second): + return "done", nil + case <-ctx.Done(): + return "", ctx.Err() + } + }) + + // Wait with a short timeout. + waitCtx, waitCancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer waitCancel() + + _, err := f.Wait(waitCtx) + if err == nil { + t.Fatal("expected context deadline exceeded") + } + + // Clean up: cancel the goroutine's context so it exits promptly. + goCancel() + rt.Wait() +} + +func TestFutureDone(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + f := rt.Go(ctx, "quick", func(ctx context.Context) (string, error) { + return "ok", nil + }) + + // Wait for completion via Done channel. + select { + case <-f.Done(): + // OK + case <-time.After(2 * time.Second): + t.Fatal("Future.Done did not close") + } +} + +func TestWait(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + var count atomic.Int32 + + for i := 0; i < 5; i++ { + rt.Go(ctx, "task", func(ctx context.Context) (string, error) { + time.Sleep(50 * time.Millisecond) + count.Add(1) + return "done", nil + }) + } + + rt.Wait() + + if count.Load() != 5 { + t.Errorf("count = %d, want 5", count.Load()) + } +} + +func TestMultipleSpawns(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + // Spawn several children of different types. + _, err := rt.SpawnAgent(ctx, newTestAgent("agent1", "a1-out"), "input") + if err != nil { + t.Fatalf("SpawnAgent error: %v", err) + } + + _, err = rt.SpawnAgent(ctx, newTestAgent("agent2", "a2-out"), "input") + if err != nil { + t.Fatalf("SpawnAgent error: %v", err) + } + + res := rt.Result() + if len(res.Children) != 2 { + t.Fatalf("Children = %d, want 2", len(res.Children)) + } + if res.Children[0].Output != "a1-out" { + t.Errorf("child[0] output = %q, want %q", res.Children[0].Output, "a1-out") + } + if res.Children[1].Output != "a2-out" { + t.Errorf("child[1] output = %q, want %q", res.Children[1].Output, "a2-out") + } +} + +func TestResultAggregation(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + // Spawn two agents. + _, _ = rt.SpawnAgent(ctx, newTestAgent("a1", "out1"), "input") + _, _ = rt.SpawnAgent(ctx, newTestAgent("a2", "out2"), "input") + + res := rt.Result() + if res.RunID == "" { + t.Error("RunID should not be empty") + } + if len(res.Children) != 2 { + t.Fatalf("Children = %d, want 2", len(res.Children)) + } + // Usage should be aggregated from both agents. + if res.TotalUsage.PromptTokens != 20 { + t.Errorf("TotalUsage.PromptTokens = %d, want 20", res.TotalUsage.PromptTokens) + } + if res.TotalUsage.CompletionTokens != 10 { + t.Errorf("TotalUsage.CompletionTokens = %d, want 10", res.TotalUsage.CompletionTokens) + } + if res.TotalUsage.TotalTokens != 30 { + t.Errorf("TotalUsage.TotalTokens = %d, want 30", res.TotalUsage.TotalTokens) + } +} + +func TestRemainingBudget(t *testing.T) { + // No budget → returns -1. + rt := New("rt") + if rt.RemainingBudget() != -1 { + t.Errorf("RemainingBudget = %f, want -1", rt.RemainingBudget()) + } + + // With budget. + rt2 := New("rt2", WithConfig(Config{CostBudget: 1.00})) + if rt2.RemainingBudget() != 1.00 { + t.Errorf("RemainingBudget = %f, want 1.00", rt2.RemainingBudget()) + } + + // After spending. + rt2.mu.Lock() + rt2.totalCost = 0.75 + rt2.mu.Unlock() + if rt2.RemainingBudget() != 0.25 { + t.Errorf("RemainingBudget = %f, want 0.25", rt2.RemainingBudget()) + } +} + +func TestTraceSpans(t *testing.T) { + tracer := trace.NewInMemory() + rt := New("rt", WithTracer(tracer)) + ctx := context.Background() + + _, _ = rt.SpawnAgent(ctx, newTestAgent("traced", "out"), "input") + + spans := tracer.Spans() + found := false + for _, s := range spans { + if s.Name == "dynamic.spawn_agent" { + found = true + if s.Attributes["dynamic.child.name"] != "traced" { + t.Errorf("child name attr = %q, want %q", + s.Attributes["dynamic.child.name"], "traced") + } + if s.Attributes["dynamic.child.type"] != "agent" { + t.Errorf("child type attr = %q, want %q", + s.Attributes["dynamic.child.type"], "agent") + } + } + } + if !found { + t.Error("dynamic.spawn_agent span not found") + } +} + +func TestTraceSpanGo(t *testing.T) { + tracer := trace.NewInMemory() + rt := New("rt", WithTracer(tracer)) + ctx := context.Background() + + f := rt.Go(ctx, "bg-task", func(ctx context.Context) (string, error) { + return "done", nil + }) + _, _ = f.Wait(ctx) + + spans := tracer.Spans() + found := false + for _, s := range spans { + if s.Name == "dynamic.go" { + found = true + if s.Attributes["dynamic.child.name"] != "bg-task" { + t.Errorf("child name = %q, want %q", + s.Attributes["dynamic.child.name"], "bg-task") + } + } + } + if !found { + t.Error("dynamic.go span not found") + } +} + +func TestParallelSpawns(t *testing.T) { + rt := New("rt", WithConfig(Config{MaxConcurrent: 3})) + ctx := context.Background() + + // Launch 3 concurrent agents via Go. + var futures []*Future + for i := 0; i < 3; i++ { + a := newTestAgent("parallel", "parallel-out") + f := rt.Go(ctx, "parallel", func(ctx context.Context) (string, error) { + r, err := rt.SpawnAgent(ctx, a, "input") + if err != nil { + return "", err + } + return r.Message.Content, nil + }) + futures = append(futures, f) + } + + for _, f := range futures { + out, err := f.Wait(ctx) + if err != nil { + t.Fatalf("parallel wait error: %v", err) + } + if out != "parallel-out" { + t.Errorf("parallel output = %q, want %q", out, "parallel-out") + } + } + + res := rt.Result() + if len(res.Children) != 3 { + t.Errorf("Children = %d, want 3", len(res.Children)) + } +} + +func TestChildContextIncreasesDepth(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + // Depth starts at 0. + childCtx := rt.childContext(ctx) + if DepthFromContext(childCtx) != 1 { + t.Errorf("child depth = %d, want 1", DepthFromContext(childCtx)) + } + + // Nested. + grandchildCtx := rt.childContext(childCtx) + if DepthFromContext(grandchildCtx) != 2 { + t.Errorf("grandchild depth = %d, want 2", DepthFromContext(grandchildCtx)) + } +} + +func TestSpawnAllTypes(t *testing.T) { + rt := New("rt") + ctx := context.Background() + + // Agent. + _, err := rt.SpawnAgent(ctx, newTestAgent("a", "agent-out"), "input") + if err != nil { + t.Fatalf("SpawnAgent: %v", err) + } + + // Team. + _, err = rt.SpawnTeam(ctx, newTestTeam("t", "team-out"), "input") + if err != nil { + t.Fatalf("SpawnTeam: %v", err) + } + + // Pipeline. + _, err = rt.SpawnPipeline(ctx, newTestPipeline("p"), "input") + if err != nil { + t.Fatalf("SpawnPipeline: %v", err) + } + + // Graph. + _, err = rt.SpawnGraph(ctx, newTestGraph("g"), "input") + if err != nil { + t.Fatalf("SpawnGraph: %v", err) + } + + res := rt.Result() + if len(res.Children) != 4 { + t.Fatalf("Children = %d, want 4", len(res.Children)) + } + + types := map[string]bool{} + for _, c := range res.Children { + types[c.Type] = true + } + for _, typ := range []string{"agent", "team", "pipeline", "graph"} { + if !types[typ] { + t.Errorf("missing child type %q", typ) + } + } +} diff --git a/website/app/docs/page.tsx b/website/app/docs/page.tsx index 0be1f2d..68be061 100644 --- a/website/app/docs/page.tsx +++ b/website/app/docs/page.tsx @@ -25,6 +25,9 @@ const sections = [ { id: "graph", label: "Graph" }, { id: "graph-builder", label: "Graph Builder" }, { id: "graph-advanced", label: "Loops & Conditions" }, + { id: "dynamic", label: "Dynamic Orchestration" }, + { id: "dynamic-governance", label: "Resource Governance" }, + { id: "dynamic-async", label: "Async & Futures" }, { id: "tracing", label: "Tracing" }, { id: "cost-tracking", label: "Cost Tracking" }, ]; @@ -133,7 +136,10 @@ export default function DocsPage() { │ ├── trace/ # Tracing and observability │ ├── cost/ # Cost tracking and budgets │ └── orchestrator/ -│ └── team/ # Team (chat room) orchestrator +│ ├── team/ # Team (chat room) orchestrator +│ ├── pipeline/ # Pipeline (linear) orchestrator +│ ├── graph/ # Graph orchestrator +│ └── dynamic/ # Dynamic orchestration runtime ├── internal/ │ └── id/ # ID generation └── cmd/ @@ -1047,6 +1053,167 @@ graph.Always()`} /> + {/* Dynamic Orchestration */} +
+

+ Dynamic Orchestration is GoGrid's most powerful pattern. A Runtime enables + agents to spawn child agents, teams, pipelines, or graphs at runtime — the executing + agent decides which orchestration to use based on the problem at hand. +

+ +

Spawning Children

+

+ Four spawn methods correspond to GoGrid's four orchestration patterns. Each + blocks until the child completes, inherits the parent's tracing context, + and records cost/usage metrics. +

+ +

Context Propagation

+

+ The runtime is stored in context so nested orchestrations can dynamically + spawn further children up to the configured depth limit. +

+ +
+ + {/* Resource Governance */} +
+

+ The runtime enforces resource limits to prevent runaway costs, infinite recursion, + and resource exhaustion. +

+
+ + + +
+ +

Cascading Cancellation

+

+ All children use derived contexts. Canceling the parent context automatically + cancels all running children — no orphaned goroutines or wasted LLM calls. +

+ +
+ + {/* Async & Futures */} +
+

+ Use Go to launch children in the background. Returns a Future that + can be awaited or polled. +

+ +

Future API

+ +
+ {/* Tracing */}

@@ -1140,6 +1307,18 @@ for _, span := range tracer.Spans() { // └─ agent.run ...`} filename="graph trace tree" /> +

Dynamic Trace Spans

+
{/* Cost Tracking */} diff --git a/website/app/examples/page.tsx b/website/app/examples/page.tsx index 8ef5415..6307ba3 100644 --- a/website/app/examples/page.tsx +++ b/website/app/examples/page.tsx @@ -49,6 +49,11 @@ const examples = [ title: "Pipeline State Transfer", desc: "State ownership enforced across pipeline stages with audit trail.", }, + { + id: "dynamic-research", + title: "Dynamic Research Coordinator", + desc: "An agent dynamically spawns teams, pipelines, and sub-agents at runtime.", + }, ]; export default function ExamplesPage() { @@ -798,6 +803,136 @@ func main() { fmt.Printf(" %s -> %s (generation %d)\\n", entry.From, entry.To, entry.Generation) } +}`} + /> + + + {/* Dynamic Research Coordinator */} + + diff --git a/website/components/Architecture.tsx b/website/components/Architecture.tsx index 213330f..8de6548 100644 --- a/website/components/Architecture.tsx +++ b/website/components/Architecture.tsx @@ -31,7 +31,7 @@ const patterns = [ name: "Dynamic Orchestration", icon: "● → ✱", description: - "Agents spawn child agents, teams, pipelines, or graphs at runtime. Unlimited scaling, minimal assumptions.", + "A Runtime enables agents to spawn child agents, teams, pipelines, or graphs at runtime. Resource governance, async futures, and aggregate metrics.", }, ]; From c81c41d7e9e850e7871fa41e06d24e6a3380ddef Mon Sep 17 00:00:00 2001 From: lonestarx1 Date: Mon, 16 Feb 2026 19:19:42 +0900 Subject: [PATCH 2/2] Remove unused costProvider type in dynamic tests --- pkg/orchestrator/dynamic/runtime_test.go | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/pkg/orchestrator/dynamic/runtime_test.go b/pkg/orchestrator/dynamic/runtime_test.go index 8537989..7575d68 100644 --- a/pkg/orchestrator/dynamic/runtime_test.go +++ b/pkg/orchestrator/dynamic/runtime_test.go @@ -57,20 +57,6 @@ func (e *errorProvider) Complete(_ context.Context, _ llm.Params) (*llm.Response return nil, e.err } -// costProvider returns a response with configurable cost. -type costProvider struct { - content string - usage llm.Usage -} - -func (c *costProvider) Complete(_ context.Context, _ llm.Params) (*llm.Response, error) { - return &llm.Response{ - Message: llm.NewAssistantMessage(c.content), - Usage: c.usage, - Model: "mock-model", - }, nil -} - // --- Helpers --- func newTestAgent(name, content string) *agent.Agent {