ideas_from_ingredients skill and job (Feature 18) (#144)

Co-authored-by: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com>

internal/recipes/ground.go

@@ -292,7 +292,22 @@ type GroundResult struct {
 // affect the others.
 func (g *Grounder) GroundMany(ctx context.Context, reqs []GroundRequest) []GroundResult {
 	results := make([]GroundResult, len(reqs))
+	g.GroundEach(ctx, reqs, func(i int, r GroundResult) { results[i] = r })
+	return results
+}
+
+// GroundEach grounds each request as GroundMany does, but calls done with
+// each item's index and result as soon as that item finishes, one call at
+// a time. It returns once every item has finished and every call has
+// returned.
+func (g *Grounder) GroundEach(ctx context.Context, reqs []GroundRequest, done func(i int, r GroundResult)) {
 	slots := make(chan struct{}, GroundConcurrency)
+	var mu sync.Mutex
+	report := func(i int, r GroundResult) {
+		mu.Lock()
+		defer mu.Unlock()
+		done(i, r)
+	}
 	var wg sync.WaitGroup
 	for i, req := range reqs {
 		wg.Add(1)
@@ -301,16 +316,15 @@ func (g *Grounder) GroundMany(ctx context.Context, reqs []GroundRequest) []Groun
 			select {
 			case slots <- struct{}{}:
 			case <-ctx.Done():
-				results[i] = GroundResult{Err: ctx.Err()}
+				report(i, GroundResult{Err: ctx.Err()})
 				return
 			}
-			defer func() { <-slots }()
 			gr, err := g.Ground(ctx, req.DishName, req.Query)
-			results[i] = GroundResult{Grounded: gr, Err: err}
+			<-slots
+			report(i, GroundResult{Grounded: gr, Err: err})
 		}()
 	}
 	wg.Wait()
-	return results
 }
 
 // stored returns the grounding of the recipe stored for sourceURL, and

internal/recipes/recipestest/recipestest.go

@@ -23,12 +23,16 @@ import (
 )
 
 // Search is a recipes.SearchSource and recipes.SearchProvider answering
-// each query from Results. Err, when set, fails every search. Gate, when
-// set, makes every search wait for it to close.
+// each query from Results. Err, when set, fails every search, and Fail
+// fails the searches for its queries. Gate, when set, makes every search
+// wait for it to close, and Gates makes the searches for its queries wait
+// for theirs.
 type Search struct {
 	Results map[string][]recipes.Candidate
 	Err     error
+	Fail    map[string]error
 	Gate    chan struct{}
+	Gates   map[string]chan struct{}
 
 	mu      sync.Mutex
 	queries []string
@@ -40,9 +44,12 @@ func (s *Search) Search(ctx context.Context, query string, n int) ([]recipes.Can
 	s.mu.Lock()
 	s.queries = append(s.queries, query)
 	s.mu.Unlock()
-	if s.Gate != nil {
+	for _, gate := range []chan struct{}{s.Gate, s.Gates[query]} {
+		if gate == nil {
+			continue
+		}
 		select {
-		case <-s.Gate:
+		case <-gate:
 		case <-ctx.Done():
 			return nil, ctx.Err()
 		}
@@ -50,6 +57,9 @@ func (s *Search) Search(ctx context.Context, query string, n int) ([]recipes.Can
 	if s.Err != nil {
 		return nil, s.Err
 	}
+	if err := s.Fail[query]; err != nil {
+		return nil, err
+	}
 	c := s.Results[query]
 	return c[:min(n, len(c))], nil
 }