Fail empty ideas jobs, clear ideas on failure, show each idea as it grounds
- A run that keeps no idea fails with no_ideas_found instead of ending
done with an empty list.
- Before an ideas job fails it republishes its result with no ideas, so
GET /api/v1/jobs/{id} never shows ideas on a failed job, even ones a
previous grounding round already published.
- Each idea is published as soon as its grounding finishes, placed in the
model's order. recipes.GroundEach reports items as they finish;
GroundMany is rebuilt on it with unchanged semantics.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user