nodes_supervisor

var Serve = action.New("supervisor.serve", func(ctx context.Context, req ServeReq) (any, error) {
	if ctx == nil {
		return nil, xerr.BadRequest("supervisor.serve: context is nil")
	}

	nodes := make([]serviceNode, len(req.Services))
	for i, service := range req.Services {
		enabled := service.Enabled == nil || *service.Enabled
		nodes[i] = serviceNode{
			id:        service.ID,
			action:    service.Action,
			dependsOn: service.DependsOn,
			enabled:   enabled,
			payload:   service.Payload,
		}
	}
	plan, err := buildServicePlan(nodes)
	if err != nil {
		return nil, xerr.BadRequest("supervisor.serve: " + err.Error())
	}

	resolver := contracts.ActionResolverFromContext(ctx)
	if resolver == nil {
		return nil, xerr.Internal("supervisor.serve: no action resolver in context")
	}
	actions := make([]action.AnyAction, len(plan))
	enabledCount := 0
	for i, service := range plan {
		act, found := resolver.Action(service.action)
		if !found || act == nil {
			return nil, xerr.NotFound("supervisor.serve: action " + service.action + " not found")
		}
		actions[i] = act
		if service.enabled {
			enabledCount++
		}
	}
	if enabledCount == 0 {
		return nil, xerr.BadRequest("supervisor.serve: no enabled services")
	}
	if err := ctx.Err(); err != nil {
		return nil, nil
	}

	ready := make([]*contracts.ReadySignal, len(plan))
	for i := range ready {
		ready[i] = contracts.NewReadySignal()
	}
	group, groupCtx := errgroup.WithContext(ctx)
	for i := range plan {
		if !plan[i].enabled {
			continue
		}
		current, act, signal := plan[i], actions[i], ready[i]
		group.Go(func() (serviceErr error) {
			defer func() {
				signal.MarkStopped()
				if recovered := recover(); recovered != nil {
					serviceErr = fmt.Errorf("service %q panicked: %v", current.id, recovered)
				}
			}()

			for _, dependency := range current.dependencies {
				select {
				case <-ready[dependency].Done():
				case <-groupCtx.Done():
					return nil
				}
			}
			if groupCtx.Err() != nil {
				return nil
			}

			serviceCtx := contracts.WithServiceReady(groupCtx, signal)
			_, invokeErr := action.InvokeAny(serviceCtx, act, current.payload)
			if ctx.Err() != nil {
				return nil
			}
			if invokeErr == nil {
				return fmt.Errorf("service %q exited unexpectedly without an error", current.id)
			}
			return fmt.Errorf("service %q exited unexpectedly: %w", current.id, invokeErr)
		})
	}

	return nil, group.Wait()
}).
	Description("Run long-lived services with readiness dependencies and sibling cancellation").
	Tag("supervisor", "serve").
	Build()

var Supervisor = action.New("supervisor.run", func(ctx context.Context, req SupervisorReq) (SupervisorRes, error) {
	if len(req.Tasks) == 0 {
		return SupervisorRes{}, xerr.BadRequest("supervisor: no child tasks provided")
	}

	compiler := contracts.CompilerFromContext(ctx)
	if compiler == nil {
		return SupervisorRes{}, xerr.Internal("supervisor: no pipeline compiler in context")
	}

	results := make([]ChildResult, len(req.Tasks))
	group, groupCtx := errgroup.WithContext(ctx)

	for i, task := range req.Tasks {
		index, current := i, task
		group.Go(func() error {
			results[index] = runChild(groupCtx, compiler, current)
			return nil
		})
	}
	if err := group.Wait(); err != nil {
		return SupervisorRes{}, err
	}

	var succeeded, failed int
	for i := range results {
		r := &results[i]
		if r.Error != "" {
			failed++
		} else {
			succeeded++
		}
	}

	return SupervisorRes{
		Total:     len(results),
		Succeeded: succeeded,
		Failed:    failed,
		Results:   results,
	}, nil
}).Description("Compile and run named child pipelines concurrently").
	Build()