Compare commits

..
Author SHA1 Message Date
dev_agentandClaude Opus 5.5 6a7c040657 Build the run's provider from the key it redacts; record render failures
CI / ci (pull_request) Successful in 1m33s
CI / ci (push) Successful in 1m33s
Review findings on b0cf6fb:

1. Config took a ready-made Provider and a separate, optional APIKey used
   only for redaction, so a provider holding a decrypted key could run
   with APIKey empty and a reply quoting that key unlabelled was stored.
   Config now holds the household's llm.Settings, the decrypted APIKey and
   a NewProvider constructor (presets.FromSettings in production). Run
   builds the provider from exactly that key and redacts the same key, so
   the two cannot differ. ProviderName, Model and Temperature come from
   Settings. A new test runs presets.FromSettings against a server that
   echoes the bearer key unlabelled in the reply text, reply JSON and
   model name, across a repair turn, and asserts the stored row never
   holds it; with redaction disabled the test fails.

2. A render failure wrote no ai_runs row. Every run with a valid Config
   now writes one: a render failure is bad_output and a provider that
   cannot be built is provider_error, each with the error text and empty
   request and response lists. TestRunRenderError now expects exactly one
   such row. A Config error (no store, provider name or model) still
   writes none, since the row cannot be written without them.

Redaction rules, the four statuses for provider-reaching runs (one row,
both turns) and Purge are unchanged.

Maestro task 53e382d8-1db3-4555-a31d-e44cf55cc98d.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 09:43:24 -04:00
dev_agentandClaude Opus 5.5 b0cf6fb951 Record every harness run in ai_runs with redaction and purge
CI / ci (push) Successful in 1m35s
CI / ci (pull_request) Successful in 1m32s
harness.Run now writes one ai_runs row per run that reaches the provider,
covering the repair turn: skill, version, provider, model, temperature,
status (ok, repaired, bad_output, provider_error), latency, summed token
counts when reported, error text and the per-turn request and response
bodies. Every stored string is redacted first: the decrypted API key,
Authorization and API-key header values, Google Chat webhook URLs and
URLs with key-like query or fragment parameters or user info. JSON is
decoded and its strings redacted, so escapes cannot hide a credential.
A row that cannot be written fails the run.

Purge deletes rows created before a cutoff (RunRetention is 30 days),
clearing suggestion_batches and week_plans references in the same
transaction.

Migration 0006 rebuilds ai_runs with the new status CHECK and a
created_at index; a pre-harness running or error row stops it.

Maestro task 53e382d8-1db3-4555-a31d-e44cf55cc98d.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 09:32:20 -04:00
dev_agent 5d5404b3cb Name normalization and dedupe (#59)
CI / ci (push) Successful in 1m32s
Co-authored-by: dev_agent <dev_agent@cubecraftlabs.com>
2026-10-05 09:16:16 -04:00
dev_agent 6ff6443db7 Household context builder with token budget (#57)
CI / ci (push) Successful in 1m34s
Co-authored-by: dev_agent <dev_agent@cubecraftlabs.com>
2026-10-05 08:34:44 -04:00
dev_agent 942ac42b86 Layered prompt rendering from embedded templates (#55)
CI / ci (push) Successful in 1m33s
Co-authored-by: dev_agent <dev_agent@cubecraftlabs.com>
2026-10-05 08:15:35 -04:00
dev_agent 55c15b1b3a Skill harness interface and repair retry (#53)
CI / ci (push) Successful in 1m31s
Co-authored-by: dev_agent <dev_agent@cubecraftlabs.com>
2026-10-05 07:55:34 -04:00
13 changed files with 34 additions and 829 deletions
+3 -22
View File
@@ -159,28 +159,9 @@ automatic** (D-021).
is always the key the provider sends. Request and response
bodies are redacted before they are stored: the decrypted API key,
Authorization and API-key header values, Google Chat webhook URLs and URLs
with key-like query parameters are removed, inside JSON strings too; the
skill, provider and model columns are redacted the same way. Writing the
row is bounded: a run whose work ended in time must write its row before
the skill's deadline or it fails with `llm.ErrTimeout` and its row records
the timeout; a failed run's row has a further fixed bound, and if even
that write fails the run returns the write error with its own and no row
is stored. A stored row's status therefore always matches what the caller
received.
with key-like query parameters are removed, inside JSON strings too.
`harness.Purge` deletes rows older than `harness.RunRetention` (30 days),
clearing the `suggestion_batches` and `week_plans` references first. The
harness does not call it: the scheduler's daily `ai_runs` purge
(Feature 20) must call `harness.Purge` with `now - RunRetention`.
- **Concurrency guards** (`harness.Guard`, one per process): `harness.Run`
takes a scope key, and concurrent runs of one skill with one scope share a
single in-flight run and its result, error included (single-flight); runs
with different scopes are independent. A semaphore lets at most
`harness.DefaultMaxCalls` (2, set by `NewGuard`) LLM calls run at once;
waiting for a slot honours the context. Each skill declares a timeout
(`harness.DefaultTimeout`, 60 s, unless it needs another) bounding the
whole run, slot waits and repair turn included; a run past it fails with
`llm.ErrTimeout`. A caller whose context ends gets the context error; the
shared run is cancelled, freeing its slot, once no caller waits for it.
clearing the `suggestion_batches` and `week_plans` references first.
- **Prompt layering** (D-007): each skill version's prompt is a
`text/template` file embedded from `ai/skills/prompts/`
(`<skill>.v<version>.tmpl`). The system prompt is always, in order:
@@ -230,7 +211,7 @@ purge are fixed and not configurable.
| `day_of_reminder` | Daily 16:30 | Tonight's dinner reminder. |
| `rotation_check` | Daily 09:00 | Asks "Keep it in rotation?" for tried dinners. |
| `week_rollover` | Monday 00:05 (fixed) | Freezes last week's Supply run into the Supply archive. |
| `ai_runs` purge | Daily (fixed) | Calls `harness.Purge` to delete `ai_runs` rows past 30-day retention (Feature 20). |
| `ai_runs` purge | Daily (fixed) | Deletes `ai_runs` rows past 30-day retention. |
Runs are recorded in `schedule_runs` and executed through `jobs`.
-3
View File
@@ -4,7 +4,6 @@ import (
"encoding/json"
"errors"
"strings"
"time"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm"
)
@@ -54,5 +53,3 @@ func (echoSkill) PostProcess(out echoOutput) (echoOutput, error) {
out.Echo = strings.TrimSpace(out.Echo)
return out, nil
}
func (echoSkill) Timeout() time.Duration { return DefaultTimeout }
-133
View File
@@ -1,133 +0,0 @@
package harness
import (
"context"
"fmt"
"sync"
"time"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm"
)
// DefaultMaxCalls is how many LLM calls a Guard lets run at once unless it
// is built with another limit.
const DefaultMaxCalls = 2
// DefaultTimeout is the run timeout a skill declares unless it needs
// another.
const DefaultTimeout = 60 * time.Second
// Guard bounds the process's runs. Runs of one skill with one scope share a
// single in-flight execution and its result (single-flight), and at most
// its limit of LLM calls run at once across every run (the semaphore).
// Build one with NewGuard and share it between every run of the process.
type Guard struct {
slots chan struct{}
mu sync.Mutex
flights map[flightKey]*flight
}
// flightKey identifies the runs that share one execution.
type flightKey struct {
skill string
scope string
}
// flight is one in-flight execution and the callers waiting for it.
type flight struct {
done chan struct{}
val any
err error
// waiters counts the callers still waiting; when the last one gives
// up, cancel stops the execution.
waiters int
cancel context.CancelFunc
}
// NewGuard returns a Guard that lets maxCalls LLM calls run at once.
func NewGuard(maxCalls int) (*Guard, error) {
if maxCalls < 1 {
return nil, fmt.Errorf("harness: guard: max concurrent LLM calls %d is not positive", maxCalls)
}
return &Guard{slots: make(chan struct{}, maxCalls), flights: map[flightKey]*flight{}}, nil
}
// do returns the result of fn for key, starting fn unless a call for key is
// already in flight, in which case it waits for that call's result
// instead. fn runs on a context that keeps the first caller's values but
// not its cancellation, so one caller giving up does not fail the others.
// When ctx is done first, do returns its error; when no caller is left
// waiting, fn's context is cancelled and do returns only once fn has
// ended, so the abandoned execution has released its slot by then.
func (g *Guard) do(ctx context.Context, key flightKey, fn func(context.Context) (any, error)) (any, error) {
g.mu.Lock()
f, ok := g.flights[key]
if !ok {
fctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
f = &flight{done: make(chan struct{}), cancel: cancel}
g.flights[key] = f
go func() {
defer cancel()
f.val, f.err = fn(fctx)
g.forget(key, f)
close(f.done)
}()
}
f.waiters++
g.mu.Unlock()
select {
case <-f.done:
return f.val, f.err
case <-ctx.Done():
}
g.mu.Lock()
f.waiters--
last := f.waiters == 0
if last {
// No caller may join an execution that is being abandoned.
g.forgetLocked(key, f)
f.cancel()
}
g.mu.Unlock()
if last {
<-f.done
}
return nil, ctx.Err()
}
// forget removes f from the in-flight executions if it is still there.
func (g *Guard) forget(key flightKey, f *flight) {
g.mu.Lock()
defer g.mu.Unlock()
g.forgetLocked(key, f)
}
func (g *Guard) forgetLocked(key flightKey, f *flight) {
if g.flights[key] == f {
delete(g.flights, key)
}
}
// limited is a Provider whose Generate calls hold one of its Guard's slots.
type limited struct {
llm.Provider
slots chan struct{}
}
// Generate waits for a free slot, or for ctx to be done, then sends req
// and frees the slot when the call returns.
func (p limited) Generate(ctx context.Context, req llm.Request) (llm.Response, error) {
// A done context never takes a slot, even when one is free.
if err := ctx.Err(); err != nil {
return llm.Response{}, err
}
select {
case p.slots <- struct{}{}:
case <-ctx.Done():
return llm.Response{}, ctx.Err()
}
defer func() { <-p.slots }()
return p.Provider.Generate(ctx, req)
}
-421
View File
@@ -1,421 +0,0 @@
package harness
import (
"context"
"errors"
"strings"
"testing"
"time"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm"
)
// timeoutSkill is echoSkill with its own timeout.
type timeoutSkill struct {
echoSkill
timeout time.Duration
}
func (s timeoutSkill) Timeout() time.Duration { return s.timeout }
// slowSkill is timeoutSkill whose PostProcess succeeds only after its
// timeout has passed.
type slowSkill struct {
timeoutSkill
}
func (s slowSkill) PostProcess(out echoOutput) (echoOutput, error) {
time.Sleep(3 * s.timeout)
return s.timeoutSkill.PostProcess(out)
}
// slowFailingSkill is timeoutSkill whose PostProcess fails only after its
// timeout has passed.
type slowFailingSkill struct {
timeoutSkill
}
func (s slowFailingSkill) PostProcess(echoOutput) (echoOutput, error) {
time.Sleep(3 * s.timeout)
return echoOutput{}, errors.New("cannot finish")
}
// blocked is one scripted Generate call that blocks until released.
type blocked struct {
started chan struct{}
release chan struct{}
}
// queueBlocked queues r on f as a call that blocks until released.
func queueBlocked(f *llm.Fake, r llm.GenerateResult) blocked {
b := blocked{started: make(chan struct{}), release: make(chan struct{})}
r.Started, r.Release = b.started, b.release
f.QueueGenerate(r)
return b
}
// result is what one Run returned.
type result struct {
out echoOutput
err error
}
// goRun starts Run of s with scope on cfg and returns where its result
// arrives.
func goRun(ctx context.Context, cfg Config, s Skill[echoInput, echoOutput], scope string) <-chan result {
ch := make(chan result, 1)
go func() {
out, err := Run(ctx, cfg, s, scope, echoInput{Word: "taco"})
ch <- result{out, err}
}()
return ch
}
// await returns what arrives on ch, failing the test after a second.
func await[T any](t *testing.T, ch <-chan T, what string) T {
t.Helper()
select {
case v := <-ch:
return v
case <-time.After(time.Second):
t.Fatalf("timed out waiting for %s", what)
panic("unreachable")
}
}
// waitStarted waits for b's Generate call to start.
func waitStarted(t *testing.T, b blocked, what string) {
t.Helper()
select {
case <-b.started:
case <-time.After(time.Second):
t.Fatalf("timed out waiting for %s to call Generate", what)
}
}
// assertNotStarted checks that b's Generate call has not started after a
// grace period.
func assertNotStarted(t *testing.T, b blocked, what string) {
t.Helper()
select {
case <-b.started:
t.Fatalf("%s called Generate while no slot was free", what)
case <-time.After(50 * time.Millisecond):
}
}
// waitWaiters waits until n callers wait for the run of skill with scope.
func waitWaiters(t *testing.T, g *Guard, skill, scope string, n int) {
t.Helper()
deadline := time.Now().Add(time.Second)
for {
g.mu.Lock()
got := 0
if f := g.flights[flightKey{skill: skill, scope: scope}]; f != nil {
got = f.waiters
}
g.mu.Unlock()
if got == n {
return
}
if time.Now().After(deadline) {
t.Fatalf("%d callers wait for %s/%s, want %d", got, skill, scope, n)
}
time.Sleep(time.Millisecond)
}
}
func TestRunSameScopeSharesOneExecution(t *testing.T) {
for name, script := range map[string]llm.GenerateResult{
"output": reply(`{"echo":"taco"}`),
"error": {Err: llm.ErrAuth},
} {
t.Run(name, func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
b := queueBlocked(&f, script)
first := goRun(context.Background(), cfg, echoSkill{}, "week-1")
waitStarted(t, b, "the first run")
second := goRun(context.Background(), cfg, echoSkill{}, "week-1")
waitWaiters(t, cfg.Guard, "echo", "week-1", 2)
close(b.release)
r1, r2 := await(t, first, "the first run"), await(t, second, "the second run")
if r1 != r2 {
t.Fatalf("results differ: %+v and %+v", r1, r2)
}
if name == "output" && (r1.err != nil || r1.out.Echo != "taco") {
t.Fatalf("result = %+v, want taco", r1)
}
if name == "error" && !errors.Is(r1.err, llm.ErrAuth) {
t.Fatalf("err = %v, want ErrAuth", r1.err)
}
assertCalls(t, &f, 1)
if rows := runRows(t, cfg.Store); len(rows) != 1 {
t.Fatalf("%d ai_runs rows, want 1 for the shared run", len(rows))
}
})
}
}
func TestRunDifferentScopesRunIndependently(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
b1 := queueBlocked(&f, reply(`{"echo":"taco"}`))
b2 := queueBlocked(&f, reply(`{"echo":"taco"}`))
week1 := goRun(context.Background(), cfg, echoSkill{}, "week-1")
week2 := goRun(context.Background(), cfg, echoSkill{}, "week-2")
// Both calls are in Generate at once: neither waits for the other.
waitStarted(t, b1, "a run")
waitStarted(t, b2, "the other run")
close(b1.release)
close(b2.release)
for _, ch := range []<-chan result{week1, week2} {
if r := await(t, ch, "a run"); r.err != nil || r.out.Echo != "taco" {
t.Fatalf("result = %+v", r)
}
}
assertCalls(t, &f, 2)
}
func TestRunThirdCallWaitsForSlot(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
b1 := queueBlocked(&f, reply(`{"echo":"taco"}`))
b2 := queueBlocked(&f, reply(`{"echo":"taco"}`))
b3 := queueBlocked(&f, reply(`{"echo":"taco"}`))
first := goRun(context.Background(), cfg, echoSkill{}, "week-1")
waitStarted(t, b1, "the first run")
second := goRun(context.Background(), cfg, echoSkill{}, "week-2")
waitStarted(t, b2, "the second run")
third := goRun(context.Background(), cfg, echoSkill{}, "week-3")
assertNotStarted(t, b3, "the third run")
if n := len(f.Requests()); n != 2 {
t.Fatalf("Generate called %d times while both slots were taken, want 2", n)
}
close(b1.release)
waitStarted(t, b3, "the third run")
close(b2.release)
close(b3.release)
for _, ch := range []<-chan result{first, second, third} {
if r := await(t, ch, "a run"); r.err != nil || r.out.Echo != "taco" {
t.Fatalf("result = %+v", r)
}
}
assertCalls(t, &f, 3)
}
func TestRunCancelledContextReleasesSlot(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
b1 := queueBlocked(&f, reply(`{"echo":"taco"}`))
b2 := queueBlocked(&f, reply(`{"echo":"taco"}`))
b3 := queueBlocked(&f, reply(`{"echo":"taco"}`))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
first := goRun(ctx, cfg, echoSkill{}, "week-1")
waitStarted(t, b1, "the first run")
second := goRun(context.Background(), cfg, echoSkill{}, "week-2")
waitStarted(t, b2, "the second run")
third := goRun(context.Background(), cfg, echoSkill{}, "week-3")
assertNotStarted(t, b3, "the third run")
cancel()
if r := await(t, first, "the cancelled run"); !errors.Is(r.err, context.Canceled) {
t.Fatalf("cancelled run err = %v, want context.Canceled", r.err)
}
// The cancelled run's slot goes to the third run; the first call's
// release is never closed.
waitStarted(t, b3, "the third run")
close(b2.release)
close(b3.release)
for _, ch := range []<-chan result{second, third} {
if r := await(t, ch, "a run"); r.err != nil || r.out.Echo != "taco" {
t.Fatalf("result = %+v", r)
}
}
assertCalls(t, &f, 3)
var statuses []string
for _, r := range runRows(t, cfg.Store) {
statuses = append(statuses, r.Status)
}
if strings.Join(statuses, ",") != "provider_error,ok,ok" {
t.Fatalf("statuses = %v, want the cancelled run recorded as provider_error", statuses)
}
}
func TestRunCallerGivingUpLeavesSharedRun(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
b := queueBlocked(&f, reply(`{"echo":"taco"}`))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
leader := goRun(ctx, cfg, echoSkill{}, "week-1")
waitStarted(t, b, "the first run")
follower := goRun(context.Background(), cfg, echoSkill{}, "week-1")
waitWaiters(t, cfg.Guard, "echo", "week-1", 2)
cancel()
if r := await(t, leader, "the cancelled caller"); !errors.Is(r.err, context.Canceled) {
t.Fatalf("cancelled caller err = %v, want context.Canceled", r.err)
}
close(b.release)
if r := await(t, follower, "the remaining caller"); r.err != nil || r.out.Echo != "taco" {
t.Fatalf("remaining caller = %+v, want the shared run's output", r)
}
assertCalls(t, &f, 1)
}
func TestRunTimeout(t *testing.T) {
t.Run("in Generate", func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
queueBlocked(&f, reply(`{"echo":"taco"}`))
_, err := Run(context.Background(), cfg, timeoutSkill{timeout: 20 * time.Millisecond}, "week-1", echoInput{Word: "taco"})
if !errors.Is(err, llm.ErrTimeout) {
t.Fatalf("err = %v, want ErrTimeout", err)
}
assertCalls(t, &f, 1)
rows := runRows(t, cfg.Store)
if len(rows) != 1 || rows[0].Status != string(StatusProviderError) || !strings.Contains(rows[0].Error.String, llm.ErrTimeout.Error()) {
t.Fatalf("rows = %+v, want one provider_error row naming the timeout", rows)
}
})
t.Run("work that succeeds late", func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
f.QueueGenerate(reply(`{"echo":"taco"}`))
s := slowSkill{timeoutSkill{timeout: 20 * time.Millisecond}}
out, err := Run(context.Background(), cfg, s, "week-1", echoInput{Word: "taco"})
if !errors.Is(err, llm.ErrTimeout) {
t.Fatalf("err = %v, want ErrTimeout", err)
}
if out != (echoOutput{}) {
t.Fatalf("out = %+v, want no output", out)
}
assertCalls(t, &f, 1)
rows := runRows(t, cfg.Store)
if len(rows) != 1 || rows[0].Status != string(StatusProviderError) || !strings.Contains(rows[0].Error.String, llm.ErrTimeout.Error()) {
t.Fatalf("rows = %+v, want one provider_error row naming the timeout", rows)
}
})
t.Run("work that fails late", func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
f.QueueGenerate(reply(`{"echo":"taco"}`))
s := slowFailingSkill{timeoutSkill{timeout: 20 * time.Millisecond}}
_, err := Run(context.Background(), cfg, s, "week-1", echoInput{Word: "taco"})
if !errors.Is(err, llm.ErrTimeout) || !strings.Contains(err.Error(), "cannot finish") {
t.Fatalf("err = %v, want ErrTimeout and the post-process failure", err)
}
assertCalls(t, &f, 1)
// The row records the timeout the caller received, not the
// bad_output a post-process failure in time would be, and keeps
// the underlying error.
rows := runRows(t, cfg.Store)
if len(rows) != 1 || rows[0].Status != string(StatusProviderError) ||
!strings.Contains(rows[0].Error.String, llm.ErrTimeout.Error()) || !strings.Contains(rows[0].Error.String, "cannot finish") {
t.Fatalf("rows = %+v, want one provider_error row naming the timeout and the post-process failure", rows)
}
})
t.Run("row saved late", func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
f.QueueGenerate(reply(`{"echo":"taco"}`))
timeout := 100 * time.Millisecond
// Holding the store's only writer connection past the deadline
// keeps the ok row from being written in time, after the work
// itself has ended in time.
conn, err := cfg.Store.Writer.Conn(context.Background())
if err != nil {
t.Fatal(err)
}
released := time.AfterFunc(3*timeout, func() { conn.Close() })
defer released.Stop()
out, err := Run(context.Background(), cfg, timeoutSkill{timeout: timeout}, "week-1", echoInput{Word: "taco"})
if !errors.Is(err, llm.ErrTimeout) {
t.Fatalf("err = %v, want ErrTimeout", err)
}
if out != (echoOutput{}) {
t.Fatalf("out = %+v, want no output", out)
}
assertCalls(t, &f, 1)
// The row records the timeout the caller received, not the ok
// outcome the work had.
rows := runRows(t, cfg.Store)
if len(rows) != 1 || rows[0].Status != string(StatusProviderError) || !strings.Contains(rows[0].Error.String, llm.ErrTimeout.Error()) {
t.Fatalf("rows = %+v, want one provider_error row naming the timeout", rows)
}
})
t.Run("writer blocked", func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
f.QueueGenerate(reply(`{"echo":"taco"}`))
timeout := 50 * time.Millisecond
saveTimeout := runSaveTimeout
runSaveTimeout = 100 * time.Millisecond
defer func() { runSaveTimeout = saveTimeout }()
// The writer stays held until Run has returned, so neither save
// can be written.
conn, err := cfg.Store.Writer.Conn(context.Background())
if err != nil {
t.Fatal(err)
}
started := time.Now()
_, err = Run(context.Background(), cfg, timeoutSkill{timeout: timeout}, "week-1", echoInput{Word: "taco"})
elapsed := time.Since(started)
conn.Close()
if !errors.Is(err, llm.ErrTimeout) || !strings.Contains(err.Error(), "record ai_runs row") {
t.Fatalf("err = %v, want ErrTimeout and the row write failure", err)
}
if bound := timeout + runSaveTimeout + time.Second; elapsed > bound {
t.Fatalf("Run took %v with the writer blocked, want at most %v", elapsed, bound)
}
if rows := runRows(t, cfg.Store); len(rows) != 0 {
t.Fatalf("rows = %+v, want none", rows)
}
})
t.Run("waiting for a slot", func(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
cfg.Guard = testGuard(t, 1)
b := queueBlocked(&f, reply(`{"echo":"taco"}`))
holder := goRun(context.Background(), cfg, echoSkill{}, "week-1")
waitStarted(t, b, "the slot holder")
_, err := Run(context.Background(), cfg, timeoutSkill{timeout: 20 * time.Millisecond}, "week-2", echoInput{Word: "taco"})
if !errors.Is(err, llm.ErrTimeout) {
t.Fatalf("err = %v, want ErrTimeout", err)
}
close(b.release)
if r := await(t, holder, "the slot holder"); r.err != nil {
t.Fatal(r.err)
}
assertCalls(t, &f, 1)
})
}
func TestRunRefusesBadGuardInput(t *testing.T) {
if _, err := NewGuard(0); err == nil {
t.Fatal("NewGuard(0) succeeded")
}
var f llm.Fake
cfg := testConfig(t, &f)
if _, err := Run(context.Background(), cfg, echoSkill{}, "", echoInput{Word: "taco"}); err == nil || !strings.Contains(err.Error(), "no scope") {
t.Fatalf("empty scope err = %v", err)
}
if _, err := Run(context.Background(), cfg, timeoutSkill{}, "week-1", echoInput{Word: "taco"}); err == nil || !strings.Contains(err.Error(), "not positive") {
t.Fatalf("zero timeout err = %v", err)
}
assertCalls(t, &f, 0)
if rows := runRows(t, cfg.Store); len(rows) != 0 {
t.Fatalf("refused runs wrote %d ai_runs rows, want none", len(rows))
}
}
+5 -101
View File
@@ -5,8 +5,6 @@
// and on failure sends exactly one repair turn quoting the problems. If the
// repair fails too the run fails with llm.ErrBadOutput; there is no silent
// fallback (D-007). Every run is recorded in ai_runs (see Run and Purge).
// A Guard shares in-flight runs of one skill and scope, bounds concurrent
// LLM calls, and each run ends at its skill's timeout.
package harness
import (
@@ -17,7 +15,6 @@ import (
"fmt"
"io"
"strings"
"time"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
@@ -42,10 +39,6 @@ type Skill[In, Out any] interface {
// is validated again; problems found then are repaired like any other.
// Its error fails the run without a repair turn.
PostProcess(out Out) (Out, error)
// Timeout bounds a whole run, waiting for LLM call slots, the repair
// turn and writing a successful run's ai_runs row included (see Run). It is DefaultTimeout unless the skill needs
// another, and must be positive.
Timeout() time.Duration
}
// NewProvider builds the provider a run sends its requests to from the
@@ -70,9 +63,6 @@ type Config struct {
Persona string
// Store receives the run's ai_runs row. It must not be nil.
Store *store.Store
// Guard is the process's Guard, shared by every run. It must not be
// nil.
Guard *Guard
}
// validate reports a Config missing what every run needs.
@@ -86,8 +76,6 @@ func (c Config) validate() error {
return errors.New("config has no model")
case c.Store == nil:
return errors.New("config has no store")
case c.Guard == nil:
return errors.New("config has no guard")
}
return nil
}
@@ -115,20 +103,6 @@ func (e *BadOutputError) Unwrap() error { return llm.ErrBadOutput }
// output. A provider error other than llm.ErrBadOutput is returned
// unchanged and never repaired.
//
// scope names what the run is for, e.g. a week, and must not be empty.
// While a run of s with scope is in flight on cfg.Guard, another Run of s
// with the same scope does not start a run of its own: it waits for that
// run and returns its output or error, whatever its own in and cfg. So a
// scope must identify the input. Runs with different scopes, or of
// different skills, run independently. Every Generate call holds one of
// cfg.Guard's slots while it runs.
//
// A run that has not ended within s.Timeout() fails with an error matching
// llm.ErrTimeout, wrapping the error its work returned if any, and its row
// records StatusProviderError whatever that work's outcome. When ctx is done first, Run returns ctx.Err(); the run
// goes on for the callers still waiting for it, and is cancelled when none
// is left.
//
// Every run with a valid Config writes one ai_runs row when it ends,
// covering both turns, with its RunStatus and the request and response
// bodies redacted (see redact). A render failure is recorded as
@@ -137,91 +111,22 @@ func (e *BadOutputError) Unwrap() error { return llm.ErrBadOutput }
// before anything else, and writes no row since the row cannot be
// written without one. When the row cannot be written the run fails with
// that error too, even if its output was usable.
//
// Writing the row is bounded too. A run whose work ended in time must
// write its row before s.Timeout() as well; if the store's writer is not
// free by then, the run fails with llm.ErrTimeout and its row records the
// timeout instead. The row of a run that failed is written within a
// further fixed bound (runSaveTimeout); if it cannot be, the run fails with
// the write error alongside its own, and no row is stored. So Run returns
// at most that bound after s.Timeout(), and a stored row's status always
// matches what the caller received.
func Run[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], scope string, in In) (Out, error) {
func Run[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], in In) (Out, error) {
var zero Out
if err := cfg.validate(); err != nil {
return zero, fmt.Errorf("harness: skill %s v%d: %w", s.Name(), s.Version(), err)
}
if scope == "" {
return zero, fmt.Errorf("harness: skill %s v%d: no scope", s.Name(), s.Version())
log := &runLog{start: store.Now()}
out, status, err := execute(ctx, cfg, s, in, log)
if saveErr := log.save(ctx, cfg, s.Name(), s.Version(), status, err); saveErr != nil {
return zero, errors.Join(err, fmt.Errorf("harness: skill %s v%d: record ai_runs row: %w", s.Name(), s.Version(), saveErr))
}
timeout := s.Timeout()
if timeout <= 0 {
return zero, fmt.Errorf("harness: skill %s v%d: timeout %v is not positive", s.Name(), s.Version(), timeout)
}
val, err := cfg.Guard.do(ctx, flightKey{skill: s.Name(), scope: scope}, func(ctx context.Context) (any, error) {
return runOnce(ctx, cfg, s, timeout, in)
})
if err != nil {
return zero, err
}
out, ok := val.(Out)
if !ok {
return zero, fmt.Errorf("harness: skill %s v%d: scope %q is shared with a run whose output is %T", s.Name(), s.Version(), scope, val)
}
return out, nil
}
// runOnce is one execution of s on in, ended at timeout, with its ai_runs row.
func runOnce[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], timeout time.Duration, in In) (Out, error) {
var zero Out
expired := fmt.Errorf("harness: skill %s v%d: %w: no result within %v", s.Name(), s.Version(), llm.ErrTimeout, timeout)
deadline := time.Now().Add(timeout)
ctx, cancel := context.WithDeadlineCause(ctx, deadline, expired)
defer cancel()
log := &runLog{id: store.NewID(), start: store.Now()}
out, status, err := execute(ctx, cfg, s, in, log)
// Work that only ends after the deadline, such as a slow PostProcess,
// is a timeout whatever it returned, so its row says provider_error
// like the caller's error, which keeps any underlying error. The clock
// is checked as well as ctx because ctx's timer may not have fired yet.
if context.Cause(ctx) == expired || !time.Now().Before(deadline) {
if err == nil {
err = expired
} else {
err = fmt.Errorf("%w: %w", expired, err)
}
status = StatusProviderError
}
// The saves ignore the callers' cancellation, so a cancelled run is
// recorded too, but each is bounded. Work that ended in time must also
// be recorded in time: its row is written within the deadline or not at
// all, so a stored ok or repaired row always means the caller got the
// output.
if err == nil {
sctx, cancel := context.WithDeadline(context.WithoutCancel(ctx), deadline)
saveErr := log.save(sctx, cfg, s.Name(), s.Version(), status, nil)
late := sctx.Err() != nil
cancel()
if saveErr == nil {
return out, nil
}
if !late {
return zero, fmt.Errorf("harness: skill %s v%d: record ai_runs row: %w", s.Name(), s.Version(), saveErr)
}
// The store's writer was not free before the deadline, so the run
// has not ended in time.
err, status = expired, StatusProviderError
}
// A run that failed is recorded within runSaveTimeout; when it cannot
// be, the run fails with both errors.
sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runSaveTimeout)
defer cancel()
if saveErr := log.save(sctx, cfg, s.Name(), s.Version(), status, err); saveErr != nil {
return zero, errors.Join(err, fmt.Errorf("harness: skill %s v%d: record ai_runs row: %w", s.Name(), s.Version(), saveErr))
}
return zero, err
}
// execute renders the prompt, builds the provider and sends the turns,
// recording them in log. It returns the run's status alongside its result.
func execute[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], in In, log *runLog) (Out, RunStatus, error) {
@@ -234,7 +139,6 @@ func execute[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], in
if err != nil {
return zero, StatusProviderError, fmt.Errorf("harness: skill %s v%d: provider: %w", s.Name(), s.Version(), err)
}
provider = limited{Provider: provider, slots: cfg.Guard.slots}
schema := s.Schema()
temperature := cfg.Settings.Temperature
req := llm.Request{
+4 -15
View File
@@ -23,22 +23,12 @@ func testConfig(t *testing.T, f *llm.Fake) Config {
Settings: llm.Settings{Provider: "anthropic", Model: "claude-test", Temperature: 0.7},
NewProvider: func(llm.Settings, string) (llm.Provider, error) { return f, nil },
Store: emptyHousehold(t),
Guard: testGuard(t, DefaultMaxCalls),
}
}
func testGuard(t *testing.T, maxCalls int) *Guard {
t.Helper()
g, err := NewGuard(maxCalls)
if err != nil {
t.Fatal(err)
}
return g
}
func run(t *testing.T, f *llm.Fake) (echoOutput, error) {
t.Helper()
return Run(context.Background(), testConfig(t, f), echoSkill{}, "week-1", echoInput{Word: "taco"})
return Run(context.Background(), testConfig(t, f), echoSkill{}, echoInput{Word: "taco"})
}
func assertCalls(t *testing.T, f *llm.Fake, want int) []llm.Request {
@@ -225,7 +215,7 @@ func TestRunProviderErrorOnRepairPassesThrough(t *testing.T) {
func TestRunRenderError(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
_, err := Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{})
_, err := Run(context.Background(), cfg, echoSkill{}, echoInput{})
if err == nil || !strings.Contains(err.Error(), "render") {
t.Fatalf("err = %v, want a render error", err)
}
@@ -245,7 +235,7 @@ func TestRunProviderBuildError(t *testing.T) {
var f llm.Fake
cfg := testConfig(t, &f)
cfg.NewProvider = func(llm.Settings, string) (llm.Provider, error) { return nil, llm.ErrMissingKey }
_, err := Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "taco"})
_, err := Run(context.Background(), cfg, echoSkill{}, echoInput{Word: "taco"})
if !errors.Is(err, llm.ErrMissingKey) {
t.Fatalf("err = %v, want ErrMissingKey", err)
}
@@ -264,12 +254,11 @@ func TestRunConfigError(t *testing.T) {
"provider name": func(c *Config) { c.Settings.Provider = "" },
"model": func(c *Config) { c.Settings.Model = "" },
"store": func(c *Config) { c.Store = nil },
"guard": func(c *Config) { c.Guard = nil },
} {
t.Run(name, func(t *testing.T) {
cfg := full
edit(&cfg)
_, err := Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "taco"})
_, err := Run(context.Background(), cfg, echoSkill{}, echoInput{Word: "taco"})
if err == nil || !strings.Contains(err.Error(), "config has no "+name) {
t.Fatalf("err = %v, want missing %s", err, name)
}
+1 -1
View File
@@ -119,7 +119,7 @@ func TestRunSendsLayeredPrompt(t *testing.T) {
in := echoInput{Word: "Taco", Context: tacoContext}
cfg := testConfig(t, &f)
cfg.Persona = persona
if _, err := Run(context.Background(), cfg, echoSkill{}, "week-1", in); err != nil {
if _, err := Run(context.Background(), cfg, echoSkill{}, in); err != nil {
t.Fatal(err)
}
req := assertCalls(t, &f, 1)[0]
+7 -18
View File
@@ -38,17 +38,9 @@ const (
// Purge with time.Now().Add(-RunRetention).
const RunRetention = 30 * 24 * time.Hour
// runSaveTimeout bounds writing the ai_runs row of a run that has failed,
// timed out or been cancelled, so a blocked writer cannot hold Run up
// indefinitely. A variable so tests can shorten it.
var runSaveTimeout = 5 * time.Second
// Purge deletes the ai_runs rows created before olderThan and returns how
// many it deleted. The suggestion_batches and week_plans references to
// those rows are cleared first, in the same transaction.
//
// Nothing in this package calls Purge: the scheduler's daily ai_runs purge
// (Feature 20) must call it with time.Now().Add(-RunRetention).
func Purge(ctx context.Context, st *store.Store, olderThan time.Time) (int64, error) {
if olderThan.IsZero() {
return 0, errors.New("harness: purge ai_runs: zero cutoff")
@@ -75,9 +67,6 @@ func Purge(ctx context.Context, st *store.Store, olderThan time.Time) (int64, er
// runLog collects one run's turns for its ai_runs row.
type runLog struct {
// id is the row's ID, fixed for the run so that a second save attempt
// can never add a second row.
id string
start store.Timestamp
requests []loggedRequest
responses []loggedResponse
@@ -149,8 +138,8 @@ func (l *runLog) add(apiKey string, req llm.Request, resp llm.Response, err erro
l.tokensOut += int64(resp.Usage.OutputTokens)
}
// save writes the run's ai_runs row within ctx, which the caller bounds.
// Every stored string is redacted.
// save writes the run's ai_runs row. It is written even when ctx is done,
// so a cancelled run is recorded too.
func (l *runLog) save(ctx context.Context, cfg Config, skill string, version int, status RunStatus, runErr error) error {
// A run that failed before its first turn stores an empty list.
request, err := marshalRedacted(nonNil(l.requests), cfg.APIKey)
@@ -169,12 +158,12 @@ func (l *runLog) save(ctx context.Context, cfg Config, skill string, version int
errText = sql.NullString{String: redact("first reply: "+strings.Join(l.problems, "; "), cfg.APIKey), Valid: true}
}
now := store.Now()
return db.New(cfg.Store.Writer).CreateAIRun(ctx, db.CreateAIRunParams{
ID: l.id,
Skill: redact(skill, cfg.APIKey),
return db.New(cfg.Store.Writer).CreateAIRun(context.WithoutCancel(ctx), db.CreateAIRunParams{
ID: store.NewID(),
Skill: skill,
SkillVersion: strconv.Itoa(version),
Provider: redact(cfg.Settings.Provider, cfg.APIKey),
Model: redact(cfg.Settings.Model, cfg.APIKey),
Provider: cfg.Settings.Provider,
Model: cfg.Settings.Model,
Temperature: cfg.Settings.Temperature,
Status: string(status),
LatencyMs: sql.NullInt64{Int64: now.Sub(l.start.Time).Milliseconds(), Valid: true},
+6 -13
View File
@@ -128,7 +128,7 @@ func TestRunWritesOneRowWithStatus(t *testing.T) {
f.QueueGenerate(tc.script...)
cfg := testConfig(t, &f)
before := store.Now()
_, runErr := Run(context.Background(), cfg, tc.skill, "week-1", echoInput{Word: "taco"})
_, runErr := Run(context.Background(), cfg, tc.skill, echoInput{Word: "taco"})
if (runErr == nil) != (tc.status == StatusOK || tc.status == StatusRepaired) {
t.Fatalf("Run err = %v for status %s", runErr, tc.status)
}
@@ -186,7 +186,7 @@ func TestRunRowPerRun(t *testing.T) {
cfg := testConfig(t, &f)
f.QueueGenerate(reply(`{"echo":"taco"}`), reply(`{"echo":"TACO"}`), reply(`{"echo":"taco"}`), llm.GenerateResult{Err: llm.ErrAuth})
for range 3 {
_, _ = Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "taco"})
_, _ = Run(context.Background(), cfg, echoSkill{}, echoInput{Word: "taco"})
}
assertCalls(t, &f, 4)
var got []string
@@ -204,7 +204,7 @@ func TestRunRecordsCancelledRun(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
f.QueueGenerate(llm.GenerateResult{Err: context.Canceled})
if _, err := Run(ctx, cfg, echoSkill{}, "week-1", echoInput{Word: "taco"}); !errors.Is(err, context.Canceled) {
if _, err := Run(ctx, cfg, echoSkill{}, echoInput{Word: "taco"}); !errors.Is(err, context.Canceled) {
t.Fatalf("err = %v, want context.Canceled", err)
}
if rows := runRows(t, cfg.Store); len(rows) != 1 || rows[0].Status != string(StatusProviderError) {
@@ -217,7 +217,7 @@ func TestRunFailsWhenRowCannotBeWritten(t *testing.T) {
cfg := testConfig(t, &f)
cfg.Settings.Provider = "gemini" // refused by the provider CHECK
f.QueueGenerate(reply(`{"echo":"taco"}`))
_, err := Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "taco"})
_, err := Run(context.Background(), cfg, echoSkill{}, echoInput{Word: "taco"})
if err == nil || !strings.Contains(err.Error(), "record ai_runs row") {
t.Fatalf("err = %v, want the row write failure", err)
}
@@ -276,10 +276,7 @@ func TestRunRowsHoldNoSecrets(t *testing.T) {
return &f, nil
}
cfg.Persona = "Our kitchen. " + leaky
// A model value can hold a credential too: every stored
// column is redacted, not only the bodies.
cfg.Settings.Model = "model-" + seededKey + " " + seededWebhook
_, _ = Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "tacos " + seededWebhook})
_, _ = Run(context.Background(), cfg, echoSkill{}, echoInput{Word: "tacos " + seededWebhook})
assertCalls(t, &f, len(script))
rows := runRows(t, cfg.Store)
@@ -292,9 +289,6 @@ func TestRunRowsHoldNoSecrets(t *testing.T) {
t.Errorf("stored row holds %q:\n%s", s, text)
}
}
if want := "model-" + redactionMarker + " " + redactionMarker; rows[0].Model != want {
t.Errorf("stored model = %q, want %q", rows[0].Model, want)
}
for _, kept := range []string{"Our kitchen.", "Repeat: tacos", redactionMarker} {
if !strings.Contains(text, kept) {
t.Errorf("stored row lacks %q:\n%s", kept, text)
@@ -340,9 +334,8 @@ func TestRunRowsHoldNoProviderKey(t *testing.T) {
APIKey: seededKey,
NewProvider: presets.FromSettings,
Store: emptyHousehold(t),
Guard: testGuard(t, DefaultMaxCalls),
}
out, err := Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "taco"})
out, err := Run(context.Background(), cfg, echoSkill{}, echoInput{Word: "taco"})
if err != nil {
t.Fatal(err)
}
+5 -27
View File
@@ -11,9 +11,7 @@ import (
// Fake is a scripted Provider for tests. Each method returns its queued
// results in order, one per call, and records every Request Generate
// receives. A call with nothing left in its queue fails, so an unexpected
// call is never silently answered. A Generate result can block until the
// test releases it, so tests can hold calls in flight. It is safe for
// concurrent use.
// call is never silently answered. It is safe for concurrent use.
type Fake struct {
mu sync.Mutex
generate []GenerateResult
@@ -28,13 +26,6 @@ var _ Provider = (*Fake)(nil)
type GenerateResult struct {
Response Response
Err error
// Started, when set, is closed once the call has taken this result and
// recorded its Request, before it blocks on Release.
Started chan struct{}
// Release, when set, makes the call block until it is closed. When
// the call's context is done first, the call returns ctx.Err() instead
// of Response and Err, as a real provider does.
Release <-chan struct{}
}
// ListModelsResult is one scripted ListModels outcome.
@@ -64,29 +55,16 @@ func (f *Fake) QueueVerify(results ...error) {
f.verify = append(f.verify, results...)
}
// Generate records req and returns the next scripted result, blocking
// first when the result has a Release channel.
func (f *Fake) Generate(ctx context.Context, req Request) (Response, error) {
// Generate records req and returns the next scripted result.
func (f *Fake) Generate(_ context.Context, req Request) (Response, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.requests = append(f.requests, cloneRequest(req))
if len(f.generate) == 0 {
n := len(f.requests)
f.mu.Unlock()
return Response{}, fmt.Errorf("llm: fake: unscripted Generate call %d", n)
return Response{}, fmt.Errorf("llm: fake: unscripted Generate call %d", len(f.requests))
}
r := f.generate[0]
f.generate = f.generate[1:]
f.mu.Unlock()
if r.Started != nil {
close(r.Started)
}
if r.Release != nil {
select {
case <-r.Release:
case <-ctx.Done():
return Response{}, ctx.Err()
}
}
return r.Response, r.Err
}
-34
View File
@@ -136,37 +136,3 @@ func TestFakeConcurrentUse(t *testing.T) {
t.Fatalf("%d Generate results unused", g)
}
}
func TestFakeGenerateBlocksUntilReleased(t *testing.T) {
var f Fake
started, release := make(chan struct{}), make(chan struct{})
f.QueueGenerate(GenerateResult{Response: Response{Text: "done"}, Started: started, Release: release})
got := make(chan string)
go func() {
resp, err := f.Generate(context.Background(), Request{})
if err != nil {
t.Error(err)
}
got <- resp.Text
}()
<-started
select {
case text := <-got:
t.Fatalf("Generate returned %q before release", text)
case <-time.After(20 * time.Millisecond):
}
close(release)
if text := <-got; text != "done" {
t.Fatalf("Generate = %q, want the scripted reply", text)
}
}
func TestFakeBlockedGenerateHonoursContext(t *testing.T) {
var f Fake
f.QueueGenerate(GenerateResult{Response: Response{Text: "never"}, Release: make(chan struct{})})
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
defer cancel()
if _, err := f.Generate(ctx, Request{}); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("err = %v, want context.DeadlineExceeded", err)
}
}
+3 -4
View File
@@ -30,8 +30,7 @@ const (
// rate limit that names no wait, waits a jittered exponential backoff. The Request's Timeout, or DefaultTimeout, is one
// deadline for the whole call, retries and waits included; a caller's
// earlier context deadline wins. A retry whose wait would not end before
// that deadline is not made: the last error is returned at once. A wait
// cut short by cancellation or the deadline returns the context's error.
// that deadline is not made: the last error is returned at once.
// ListModels and Verify pass straight through.
type Retrying struct {
Provider
@@ -84,8 +83,8 @@ func (r *Retrying) Generate(ctx context.Context, req Request) (Response, error)
if !r.now().Add(wait).Before(deadline) {
return resp, err
}
if err := r.sleep(ctx, wait); err != nil {
return Response{}, err
if serr := r.sleep(ctx, wait); serr != nil {
return resp, err
}
}
}
-37
View File
@@ -240,43 +240,6 @@ func TestRetryingStopsWhenContextDone(t *testing.T) {
assertConsumed(t, f, 1)
}
func TestRetryingWaitInterruptedReturnsContextError(t *testing.T) {
t.Run("cancelled", func(t *testing.T) {
r, f, _ := newTestRetrying()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
r.sleep = func(ctx context.Context, d time.Duration) error {
cancel()
return sleep(ctx, d)
}
f.QueueGenerate(fail(NewRateLimitedError(3*time.Second, "")))
_, err := r.Generate(ctx, retryReq)
if !errors.Is(err, context.Canceled) || errors.Is(err, ErrRateLimited) {
t.Fatalf("err = %v, want context.Canceled", err)
}
assertConsumed(t, f, 1)
})
t.Run("deadline", func(t *testing.T) {
r, f, _ := newTestRetrying()
// The fake clock stands still, so the 500ms backoff fits before
// the caller's deadline; the real wait outlasts it.
r.jitter = func(time.Duration) time.Duration { return 0 }
r.sleep = func(ctx context.Context, _ time.Duration) error {
return sleep(ctx, time.Hour)
}
ctx, cancel := context.WithTimeout(context.Background(), 600*time.Millisecond)
defer cancel()
f.QueueGenerate(fail(NewRateLimitedError(0, "")))
_, err := r.Generate(ctx, retryReq)
if !errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrRateLimited) {
t.Fatalf("err = %v, want context.DeadlineExceeded", err)
}
assertConsumed(t, f, 1)
})
}
// slowProvider is a Provider whose every attempt takes d on clock c,
// records the deadline of the context it received, and times out.
type slowProvider struct {