Require secrets for ProviderError redaction and omit unredactable bodies

NewProviderError now takes the secrets to redact as a required []string
instead of an optional variadic. When no non-empty secret is given the
body cannot be redacted, so no excerpt is stored and Body()/Error() never
expose raw vendor output.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

cmd/server/jobs_test.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+)
+
+// TestNewJobRunnerRunsNoop proves the runner run builds can execute a job:
+// a noop job enqueued before Start, as at boot, succeeds with its result.
+func TestNewJobRunnerRunsNoop(t *testing.T) {
+	ctx := context.Background()
+	db, err := openStore(ctx, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := db.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	runner, kinds, err := newJobRunner(db, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	id, err := kinds.builtins.Noop.Enqueue(ctx, jobs.NoopPayload{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jobsCtx, stopJobs := context.WithCancel(ctx)
+	if err := runner.Start(jobsCtx); err != nil {
+		stopJobs()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		stopJobs()
+		runner.Wait()
+	})
+
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		job, err := runner.Get(ctx, id)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if job.Status == jobs.StatusSucceeded || job.Status == jobs.StatusFailed {
+			if job.Kind != jobs.KindNoop || job.Status != jobs.StatusSucceeded || string(job.Result) != `{"ok":true}` || job.Error != nil {
+				t.Fatalf("noop job = %+v (result %s), want succeeded with {\"ok\":true}", job, job.Result)
+			}
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("noop job never finished; last %+v", job)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}

cmd/server/main.go

@@ -18,6 +18,7 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/forwarded"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/frontend"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/secretbox"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/server"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
@@ -27,6 +28,9 @@ import (
 // version is set at build time with -ldflags "-X main.version=<version>".
 var version = "dev"
 
+// jobWorkers is the number of background jobs run at once.
+const jobWorkers = 2
+
 func main() {
 	cfg, err := config.Load()
 	if err != nil {
@@ -63,6 +67,10 @@ func run(cfg config.Config, logger *slog.Logger) error {
 		return err
 	}
 	settings := household.NewSettingsService(db)
+	runner, _, err := newJobRunner(db, logger)
+	if err != nil {
+		return err
+	}
 
 	sessions, err := auth.NewSessionManager(db, cfg.SessionIdle, cfg.SessionMax)
 	if err != nil {
@@ -80,10 +88,22 @@ func run(cfg config.Config, logger *slog.Logger) error {
 		return err
 	}
 
-	router, err := server.NewRouter(logger, version, cfg.TrustedProxy, authFlow, sessions, settings, frontend.Assets())
+	router, err := server.NewRouter(logger, version, cfg.TrustedProxy, authFlow, sessions, settings, runner, frontend.Assets())
 	if err != nil {
 		return err
 	}
+
+	// Start fails the jobs a previous process left running before any
+	// worker runs. The workers stop, and are waited for, before the
+	// database closes.
+	jobsCtx, stopJobs := context.WithCancel(ctx)
+	defer func() {
+		stopJobs()
+		runner.Wait()
+	}()
+	if err := runner.Start(jobsCtx); err != nil {
+		return err
+	}
 	srv := &http.Server{
 		Addr:              cfg.HTTPAddr,
 		Handler:           router,
@@ -165,6 +185,35 @@ func seedDefaults(ctx context.Context, db *store.Store, defaultTZ string) error
 	return household.SeedNotificationPrefs(ctx, db)
 }
 
+// jobKinds enqueues every job kind the server runs.
+type jobKinds struct {
+	builtins *jobs.Builtins
+}
+
+// newJobRunner returns the job runner with every job kind registered, ready
+// to Start.
+func newJobRunner(db *store.Store, logger *slog.Logger) (*jobs.Runner, *jobKinds, error) {
+	runner, err := jobs.NewRunner(db, logger, jobWorkers)
+	if err != nil {
+		return nil, nil, err
+	}
+	kinds, err := registerJobKinds(runner)
+	if err != nil {
+		return nil, nil, err
+	}
+	return runner, kinds, nil
+}
+
+// registerJobKinds registers every job kind's handler on runner. Each
+// feature's job kinds are registered here.
+func registerJobKinds(runner *jobs.Runner) (*jobKinds, error) {
+	builtins, err := jobs.RegisterBuiltins(runner)
+	if err != nil {
+		return nil, err
+	}
+	return &jobKinds{builtins: builtins}, nil
+}
+
 // logListening logs the startup line. Only the public URL's origin is logged
 // so no credential-bearing URL component can reach the log.
 func logListening(logger *slog.Logger, cfg config.Config) {

internal/ai/llm/errors.go

@@ -0,0 +1,103 @@
+package llm
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+	"unicode/utf8"
+)
+
+var (
+	// ErrAuth means the vendor rejected the credentials.
+	ErrAuth = errors.New("llm: authentication failed")
+	// ErrRateLimited means the vendor throttled the request. The concrete
+	// error is a *RateLimitedError carrying the requested wait.
+	ErrRateLimited = errors.New("llm: rate limited")
+	// ErrTimeout means the request did not finish within its timeout.
+	ErrTimeout = errors.New("llm: request timed out")
+	// ErrBadOutput means the model replied but the reply is unusable, e.g.
+	// structured output that does not parse.
+	ErrBadOutput = errors.New("llm: unusable model output")
+	// ErrProvider means the vendor returned an unexpected HTTP status. The
+	// concrete error is a *ProviderError carrying the status and a redacted
+	// body excerpt.
+	ErrProvider = errors.New("llm: provider error")
+)
+
+// RateLimitedError is a throttled request. It matches ErrRateLimited.
+type RateLimitedError struct {
+	// RetryAfter is how long the vendor asked callers to wait; 0 means it
+	// did not say.
+	RetryAfter time.Duration
+}
+
+func (e *RateLimitedError) Error() string {
+	if e.RetryAfter == 0 {
+		return ErrRateLimited.Error()
+	}
+	return fmt.Sprintf("%s: retry after %v", ErrRateLimited, e.RetryAfter)
+}
+
+// Is reports whether target is ErrRateLimited.
+func (e *RateLimitedError) Is(target error) bool { return target == ErrRateLimited }
+
+// MaxBodyExcerpt is the longest response-body excerpt, in bytes, a
+// ProviderError keeps.
+const MaxBodyExcerpt = 512
+
+// redacted replaces each secret in a ProviderError body.
+const redacted = "[REDACTED]"
+
+// ProviderError is an unexpected HTTP status from the vendor. It matches
+// ErrProvider. Build it with NewProviderError, which redacts the body.
+type ProviderError struct {
+	status int
+	body   string
+}
+
+// NewProviderError returns the error for an unexpected HTTP status.
+// secrets must hold the API key (and any other credential sent with the
+// request). Every occurrence of each non-empty secret in body is replaced
+// before the body is cut to MaxBodyExcerpt bytes, so no part of a secret
+// survives. If secrets holds no non-empty value the body cannot be
+// redacted, so no excerpt is kept at all.
+func NewProviderError(status int, body []byte, secrets []string) *ProviderError {
+	s := string(body)
+	redactedAny := false
+	for _, secret := range secrets {
+		if secret != "" {
+			s = strings.ReplaceAll(s, secret, redacted)
+			redactedAny = true
+		}
+	}
+	if !redactedAny {
+		return &ProviderError{status: status}
+	}
+	s = strings.TrimSpace(s)
+	if len(s) > MaxBodyExcerpt {
+		cut := MaxBodyExcerpt
+		for cut > 0 && !utf8.RuneStart(s[cut]) {
+			cut--
+		}
+		s = s[:cut] + "…"
+	}
+	return &ProviderError{status: status, body: s}
+}
+
+// Status returns the HTTP status code.
+func (e *ProviderError) Status() int { return e.status }
+
+// Body returns the redacted, truncated response-body excerpt; it is empty
+// when the vendor sent no body or no secret was given to redact it with.
+func (e *ProviderError) Body() string { return e.body }
+
+func (e *ProviderError) Error() string {
+	if e.body == "" {
+		return fmt.Sprintf("%s: HTTP %d", ErrProvider, e.status)
+	}
+	return fmt.Sprintf("%s: HTTP %d: %s", ErrProvider, e.status, e.body)
+}
+
+// Is reports whether target is ErrProvider.
+func (e *ProviderError) Is(target error) bool { return target == ErrProvider }

internal/ai/llm/errors_test.go

@@ -0,0 +1,121 @@
+package llm
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestSentinelErrorsMatchWhenWrapped(t *testing.T) {
+	for _, sentinel := range []error{ErrAuth, ErrTimeout, ErrBadOutput} {
+		wrapped := fmt.Errorf("openai: generate: %w", sentinel)
+		if !errors.Is(wrapped, sentinel) {
+			t.Errorf("errors.Is(%v, %v) = false", wrapped, sentinel)
+		}
+		for _, other := range []error{ErrAuth, ErrTimeout, ErrBadOutput, ErrRateLimited, ErrProvider} {
+			if other != sentinel && errors.Is(wrapped, other) {
+				t.Errorf("errors.Is(%v, %v) = true", wrapped, other)
+			}
+		}
+	}
+}
+
+func TestRateLimitedError(t *testing.T) {
+	err := fmt.Errorf("anthropic: %w", &RateLimitedError{RetryAfter: 30 * time.Second})
+	if !errors.Is(err, ErrRateLimited) {
+		t.Fatal("errors.Is(err, ErrRateLimited) = false")
+	}
+	if errors.Is(err, ErrProvider) || errors.Is(err, ErrAuth) {
+		t.Fatal("rate-limited error matches an unrelated sentinel")
+	}
+	var rl *RateLimitedError
+	if !errors.As(err, &rl) {
+		t.Fatal("errors.As(err, *RateLimitedError) = false")
+	}
+	if rl.RetryAfter != 30*time.Second {
+		t.Fatalf("RetryAfter = %v, want 30s", rl.RetryAfter)
+	}
+	if got := err.Error(); got != "anthropic: llm: rate limited: retry after 30s" {
+		t.Fatalf("Error() = %q", got)
+	}
+	if got := (&RateLimitedError{}).Error(); got != "llm: rate limited" {
+		t.Fatalf("Error() without RetryAfter = %q", got)
+	}
+}
+
+func TestProviderError(t *testing.T) {
+	err := fmt.Errorf("gemini: %w", NewProviderError(503, []byte(` {"error":"overloaded"} `), []string{"sk-test"}))
+	if !errors.Is(err, ErrProvider) {
+		t.Fatal("errors.Is(err, ErrProvider) = false")
+	}
+	if errors.Is(err, ErrRateLimited) || errors.Is(err, ErrAuth) {
+		t.Fatal("provider error matches an unrelated sentinel")
+	}
+	var pe *ProviderError
+	if !errors.As(err, &pe) {
+		t.Fatal("errors.As(err, *ProviderError) = false")
+	}
+	if pe.Status() != 503 || pe.Body() != `{"error":"overloaded"}` {
+		t.Fatalf("Status, Body = %d, %q", pe.Status(), pe.Body())
+	}
+	if got := err.Error(); got != `gemini: llm: provider error: HTTP 503: {"error":"overloaded"}` {
+		t.Fatalf("Error() = %q", got)
+	}
+	if got := NewProviderError(500, nil, []string{"sk-test"}).Error(); got != "llm: provider error: HTTP 500" {
+		t.Fatalf("Error() without body = %q", got)
+	}
+}
+
+func TestProviderErrorRedactsSecrets(t *testing.T) {
+	const key = "sk-live-0123456789abcdef"
+	body := []byte(`{"error":"invalid key sk-live-0123456789abcdef","echo":"Bearer sk-live-0123456789abcdef"}`)
+	pe := NewProviderError(400, body, []string{key, ""})
+	for _, s := range []string{pe.Error(), pe.Body()} {
+		if strings.Contains(s, key) {
+			t.Fatalf("secret leaked: %q", s)
+		}
+	}
+	want := `{"error":"invalid key [REDACTED]","echo":"Bearer [REDACTED]"}`
+	if pe.Body() != want {
+		t.Fatalf("Body() = %q, want %q", pe.Body(), want)
+	}
+}
+
+func TestProviderErrorRedactsBeforeTruncating(t *testing.T) {
+	const key = "sk-secret-straddling-the-cut"
+	// The key starts just before the excerpt limit, so truncating first
+	// would leave its prefix in the excerpt.
+	body := strings.Repeat("x", MaxBodyExcerpt-5) + key
+	pe := NewProviderError(500, []byte(body), []string{key})
+	if strings.Contains(pe.Body(), "sk-se") {
+		t.Fatalf("secret prefix leaked: %q", pe.Body())
+	}
+}
+
+func TestProviderErrorTruncatesOnRuneBoundary(t *testing.T) {
+	body := strings.Repeat("a", MaxBodyExcerpt-1) + "é" + strings.Repeat("b", 100)
+	pe := NewProviderError(500, []byte(body), []string{"sk-test"})
+	want := strings.Repeat("a", MaxBodyExcerpt-1) + "…"
+	if pe.Body() != want {
+		t.Fatalf("Body() has %d bytes, want %d", len(pe.Body()), len(want))
+	}
+}
+
+func TestProviderErrorOmitsBodyWithoutSecret(t *testing.T) {
+	const key = "sk-live-0123456789abcdef"
+	body := []byte(`{"error":"invalid key sk-live-0123456789abcdef"}`)
+	for _, secrets := range [][]string{nil, {}, {""}} {
+		pe := NewProviderError(401, body, secrets)
+		if pe.Body() != "" {
+			t.Fatalf("secrets %q: Body() = %q, want empty", secrets, pe.Body())
+		}
+		if got := pe.Error(); got != "llm: provider error: HTTP 401" {
+			t.Fatalf("secrets %q: Error() = %q", secrets, got)
+		}
+		if strings.Contains(pe.Error(), key) {
+			t.Fatalf("secrets %q: secret leaked: %q", secrets, pe.Error())
+		}
+	}
+}

internal/ai/llm/fake.go

@@ -0,0 +1,127 @@
+package llm
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"sync"
+)
+
+// 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. It is safe for concurrent use.
+type Fake struct {
+	mu       sync.Mutex
+	generate []GenerateResult
+	models   []ListModelsResult
+	verify   []error
+	requests []Request
+}
+
+var _ Provider = (*Fake)(nil)
+
+// GenerateResult is one scripted Generate outcome.
+type GenerateResult struct {
+	Response Response
+	Err      error
+}
+
+// ListModelsResult is one scripted ListModels outcome.
+type ListModelsResult struct {
+	Models []Model
+	Err    error
+}
+
+// QueueGenerate appends results to the Generate script.
+func (f *Fake) QueueGenerate(results ...GenerateResult) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.generate = append(f.generate, results...)
+}
+
+// QueueListModels appends results to the ListModels script.
+func (f *Fake) QueueListModels(results ...ListModelsResult) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.models = append(f.models, results...)
+}
+
+// QueueVerify appends results to the Verify script; a nil entry is success.
+func (f *Fake) QueueVerify(results ...error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.verify = append(f.verify, results...)
+}
+
+// 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 {
+		return Response{}, fmt.Errorf("llm: fake: unscripted Generate call %d", len(f.requests))
+	}
+	r := f.generate[0]
+	f.generate = f.generate[1:]
+	return r.Response, r.Err
+}
+
+// ListModels returns the next scripted result.
+func (f *Fake) ListModels(context.Context) ([]Model, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if len(f.models) == 0 {
+		return nil, fmt.Errorf("llm: fake: unscripted ListModels call")
+	}
+	r := f.models[0]
+	f.models = f.models[1:]
+	return slices.Clone(r.Models), r.Err
+}
+
+// Verify returns the next scripted result.
+func (f *Fake) Verify(context.Context) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if len(f.verify) == 0 {
+		return fmt.Errorf("llm: fake: unscripted Verify call")
+	}
+	err := f.verify[0]
+	f.verify = f.verify[1:]
+	return err
+}
+
+// Requests returns a copy of every Request Generate received, in order.
+func (f *Fake) Requests() []Request {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := make([]Request, len(f.requests))
+	for i, r := range f.requests {
+		out[i] = cloneRequest(r)
+	}
+	return out
+}
+
+// Remaining reports how many scripted results of each kind are unused, so a
+// test can assert its script was consumed.
+func (f *Fake) Remaining() (generate, listModels, verify int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.generate), len(f.models), len(f.verify)
+}
+
+// cloneRequest deep-copies r so a recorded Request is unaffected by later
+// changes the caller makes to its slices and pointers.
+func cloneRequest(r Request) Request {
+	r.Messages = slices.Clone(r.Messages)
+	if r.Temperature != nil {
+		t := *r.Temperature
+		r.Temperature = &t
+	}
+	if r.Schema != nil {
+		s := JSONSchema{Name: r.Schema.Name, Schema: json.RawMessage(slices.Clone([]byte(r.Schema.Schema)))}
+		r.Schema = &s
+	}
+	return r
+}

internal/ai/llm/fake_test.go

@@ -0,0 +1,138 @@
+package llm
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestFakeGenerateReturnsScriptInOrder(t *testing.T) {
+	ctx := context.Background()
+	var f Fake
+	f.QueueGenerate(
+		GenerateResult{Response: Response{Text: "first", Model: "m1"}},
+		GenerateResult{Err: &RateLimitedError{RetryAfter: time.Second}},
+	)
+	f.QueueGenerate(GenerateResult{Response: Response{JSON: json.RawMessage(`{"ok":true}`), FinishReason: "stop"}})
+
+	req := Request{Messages: []Message{{Role: RoleUser, Content: "one"}}}
+	resp, err := f.Generate(ctx, req)
+	if err != nil || resp.Text != "first" || resp.Model != "m1" {
+		t.Fatalf("call 1 = %+v, %v", resp, err)
+	}
+	if _, err := f.Generate(ctx, req); !errors.Is(err, ErrRateLimited) {
+		t.Fatalf("call 2 err = %v, want ErrRateLimited", err)
+	}
+	resp, err = f.Generate(ctx, req)
+	if err != nil || string(resp.JSON) != `{"ok":true}` || resp.FinishReason != "stop" {
+		t.Fatalf("call 3 = %+v, %v", resp, err)
+	}
+	if _, err := f.Generate(ctx, req); err == nil {
+		t.Fatal("unscripted call 4 succeeded")
+	}
+	if g, l, v := f.Remaining(); g != 0 || l != 0 || v != 0 {
+		t.Fatalf("Remaining() = %d, %d, %d", g, l, v)
+	}
+	if n := len(f.Requests()); n != 4 {
+		t.Fatalf("recorded %d requests, want 4 (unscripted calls are recorded too)", n)
+	}
+}
+
+func TestFakeRecordsRequests(t *testing.T) {
+	var f Fake
+	f.QueueGenerate(GenerateResult{}, GenerateResult{})
+	temp := 0.2
+	first := Request{
+		System:      "be brief",
+		Messages:    []Message{{Role: RoleUser, Content: "plan dinners"}},
+		Temperature: &temp,
+		MaxTokens:   100,
+		Schema:      &JSONSchema{Name: "plan", Schema: json.RawMessage(`{"type":"object"}`)},
+		Timeout:     5 * time.Second,
+	}
+	if _, err := f.Generate(context.Background(), first); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.Generate(context.Background(), Request{System: "second"}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Later changes by the caller must not alter what was recorded.
+	first.Messages[0].Content = "changed"
+	temp = 1
+	first.Schema.Schema[2] = 'X'
+
+	got := f.Requests()
+	if len(got) != 2 || got[1].System != "second" {
+		t.Fatalf("Requests() = %+v", got)
+	}
+	r := got[0]
+	if r.System != "be brief" || r.Messages[0].Content != "plan dinners" || *r.Temperature != 0.2 ||
+		r.MaxTokens != 100 || r.Schema.Name != "plan" || string(r.Schema.Schema) != `{"type":"object"}` ||
+		r.Timeout != 5*time.Second {
+		t.Fatalf("recorded request = %+v", r)
+	}
+
+	// Changing a returned copy must not alter the record either.
+	got[0].Messages[0].Content = "mutated"
+	if f.Requests()[0].Messages[0].Content != "plan dinners" {
+		t.Fatal("Requests() returned shared state")
+	}
+}
+
+func TestFakeListModelsAndVerify(t *testing.T) {
+	ctx := context.Background()
+	var f Fake
+	f.QueueListModels(
+		ListModelsResult{Models: []Model{{ID: "a", DisplayName: "A"}, {ID: "b", DisplayName: "b"}}},
+		ListModelsResult{Err: ErrAuth},
+	)
+	f.QueueVerify(nil, fmt.Errorf("wrapped: %w", ErrAuth))
+
+	models, err := f.ListModels(ctx)
+	if err != nil || len(models) != 2 || models[0].ID != "a" {
+		t.Fatalf("ListModels 1 = %+v, %v", models, err)
+	}
+	if _, err := f.ListModels(ctx); !errors.Is(err, ErrAuth) {
+		t.Fatalf("ListModels 2 err = %v, want ErrAuth", err)
+	}
+	if _, err := f.ListModels(ctx); err == nil {
+		t.Fatal("unscripted ListModels succeeded")
+	}
+	if err := f.Verify(ctx); err != nil {
+		t.Fatalf("Verify 1 = %v", err)
+	}
+	if err := f.Verify(ctx); !errors.Is(err, ErrAuth) {
+		t.Fatalf("Verify 2 = %v, want ErrAuth", err)
+	}
+	if err := f.Verify(ctx); err == nil {
+		t.Fatal("unscripted Verify succeeded")
+	}
+}
+
+func TestFakeConcurrentUse(t *testing.T) {
+	const n = 50
+	var f Fake
+	for i := range n {
+		f.QueueGenerate(GenerateResult{Response: Response{Text: fmt.Sprint(i)}})
+	}
+	var wg sync.WaitGroup
+	for range n {
+		wg.Go(func() {
+			if _, err := f.Generate(context.Background(), Request{}); err != nil {
+				t.Error(err)
+			}
+		})
+	}
+	wg.Wait()
+	if len(f.Requests()) != n {
+		t.Fatalf("recorded %d requests, want %d", len(f.Requests()), n)
+	}
+	if g, _, _ := f.Remaining(); g != 0 {
+		t.Fatalf("%d Generate results unused", g)
+	}
+}

internal/ai/llm/llm.go

@@ -0,0 +1,132 @@
+// Package llm is the single interface MealQuest uses to talk to a language
+// model. Every vendor adapter implements Provider over plain net/http (no
+// vendor SDKs) and reports failures with this package's typed errors, so
+// callers decide what to do from errors.Is and errors.As alone.
+//
+// Errors from this package and its adapters never contain an API key.
+package llm
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+)
+
+// Provider is a language-model vendor.
+type Provider interface {
+	// Generate sends one request and returns the model's reply.
+	Generate(ctx context.Context, req Request) (Response, error)
+	// ListModels returns the models the configured credentials can use.
+	ListModels(ctx context.Context) ([]Model, error)
+	// Verify checks that the configured credentials are accepted.
+	Verify(ctx context.Context) error
+}
+
+// Role is the author of a Message.
+type Role string
+
+// The roles a Message may have. The system prompt is Request.System, not a
+// message.
+const (
+	RoleUser      Role = "user"
+	RoleAssistant Role = "assistant"
+)
+
+// Message is one turn of a conversation.
+type Message struct {
+	Role    Role
+	Content string
+}
+
+// JSONSchema is a named structured-output contract: when a Request carries
+// one, the model must reply with JSON matching Schema.
+type JSONSchema struct {
+	// Name identifies the contract to the vendor, e.g. "side_quests".
+	Name string
+	// Schema is a JSON Schema object.
+	Schema json.RawMessage
+}
+
+// Request is one generation request.
+type Request struct {
+	// System is the system prompt; empty means none.
+	System string
+	// Messages is the conversation, oldest first. It must not be empty.
+	Messages []Message
+	// Temperature is the sampling temperature; nil means the vendor default.
+	Temperature *float64
+	// MaxTokens caps the reply length; 0 means the vendor default.
+	MaxTokens int
+	// Schema, when set, requests structured output; the reply is then in
+	// Response.JSON.
+	Schema *JSONSchema
+	// Timeout bounds the request; 0 means only ctx bounds it.
+	Timeout time.Duration
+}
+
+// Validate reports whether r is well formed. Adapters call it before
+// contacting the vendor.
+func (r Request) Validate() error {
+	if len(r.Messages) == 0 {
+		return errors.New("llm: request has no messages")
+	}
+	for i, m := range r.Messages {
+		if m.Role != RoleUser && m.Role != RoleAssistant {
+			return fmt.Errorf("llm: message %d has unknown role %q", i, m.Role)
+		}
+	}
+	if r.Temperature != nil && *r.Temperature < 0 {
+		return fmt.Errorf("llm: temperature %v is negative", *r.Temperature)
+	}
+	if r.MaxTokens < 0 {
+		return fmt.Errorf("llm: max tokens %d is negative", r.MaxTokens)
+	}
+	if r.Timeout < 0 {
+		return fmt.Errorf("llm: timeout %v is negative", r.Timeout)
+	}
+	if r.Schema != nil {
+		if r.Schema.Name == "" {
+			return errors.New("llm: schema has no name")
+		}
+		var obj map[string]json.RawMessage
+		if err := json.Unmarshal(r.Schema.Schema, &obj); err != nil || obj == nil {
+			return fmt.Errorf("llm: schema %q is not a JSON object", r.Schema.Name)
+		}
+	}
+	return nil
+}
+
+// Usage is the token count of one Generate call.
+type Usage struct {
+	InputTokens  int
+	OutputTokens int
+}
+
+// Total returns InputTokens + OutputTokens.
+func (u Usage) Total() int { return u.InputTokens + u.OutputTokens }
+
+// Response is the model's reply to a Request.
+type Response struct {
+	// Text is the reply text.
+	Text string
+	// JSON is the structured reply when the Request carried a Schema, and
+	// nil otherwise.
+	JSON json.RawMessage
+	// Usage is the tokens the call consumed.
+	Usage Usage
+	// Model is the model that answered, as the vendor reports it.
+	Model string
+	// FinishReason is why generation stopped, as the vendor reports it.
+	FinishReason string
+}
+
+// Model is a model a Provider offers.
+type Model struct {
+	// ID is the vendor's model identifier, used in requests.
+	ID string
+	// DisplayName is a human-readable name; it equals ID when the vendor
+	// gives none.
+	DisplayName string
+}

internal/ai/llm/llm_test.go

@@ -0,0 +1,44 @@
+package llm
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestRequestValidate(t *testing.T) {
+	neg := -0.1
+	valid := func() Request {
+		return Request{
+			Messages: []Message{{Role: RoleUser, Content: "hi"}, {Role: RoleAssistant, Content: "hello"}},
+			Schema:   &JSONSchema{Name: "plan", Schema: json.RawMessage(`{"type":"object"}`)},
+		}
+	}
+	if err := valid().Validate(); err != nil {
+		t.Fatalf("valid request: %v", err)
+	}
+	cases := map[string]func(*Request){
+		"no messages":    func(r *Request) { r.Messages = nil },
+		"unknown role":   func(r *Request) { r.Messages[0].Role = "system" },
+		"negative temp":  func(r *Request) { r.Temperature = &neg },
+		"negative max":   func(r *Request) { r.MaxTokens = -1 },
+		"negative time":  func(r *Request) { r.Timeout = -1 },
+		"unnamed schema": func(r *Request) { r.Schema.Name = "" },
+		"schema array":   func(r *Request) { r.Schema.Schema = json.RawMessage(`[]`) },
+		"schema null":    func(r *Request) { r.Schema.Schema = json.RawMessage(`null`) },
+		"schema invalid": func(r *Request) { r.Schema.Schema = json.RawMessage(`{`) },
+		"schema empty":   func(r *Request) { r.Schema.Schema = nil },
+	}
+	for name, mutate := range cases {
+		r := valid()
+		mutate(&r)
+		if err := r.Validate(); err == nil {
+			t.Errorf("%s: Validate() = nil", name)
+		}
+	}
+}
+
+func TestUsageTotal(t *testing.T) {
+	if got := (Usage{InputTokens: 12, OutputTokens: 30}).Total(); got != 42 {
+		t.Fatalf("Total() = %d, want 42", got)
+	}
+}

internal/db/jobs.sql.go

@@ -0,0 +1,166 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.31.1
+// source: jobs.sql
+
+package db
+
+import (
+	"context"
+	"database/sql"
+)
+
+const claimNextJob = `-- name: ClaimNextJob :one
+UPDATE jobs
+SET status = 'running', started_at = ?1, updated_at = ?1
+WHERE id = (SELECT id FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 1)
+RETURNING id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at
+`
+
+// Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
+// when no job is queued.
+func (q *Queries) ClaimNextJob(ctx context.Context, now sql.NullString) (Job, error) {
+	row := q.db.QueryRowContext(ctx, claimNextJob, now)
+	var i Job
+	err := row.Scan(
+		&i.ID,
+		&i.Kind,
+		&i.Status,
+		&i.Payload,
+		&i.Result,
+		&i.Error,
+		&i.StartedAt,
+		&i.FinishedAt,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
+const createJob = `-- name: CreateJob :exec
+INSERT INTO jobs (id, kind, status, payload, created_at, updated_at)
+VALUES (?, ?, 'queued', ?, ?, ?)
+`
+
+type CreateJobParams struct {
+	ID        string
+	Kind      string
+	Payload   string
+	CreatedAt string
+	UpdatedAt string
+}
+
+func (q *Queries) CreateJob(ctx context.Context, arg CreateJobParams) error {
+	_, err := q.db.ExecContext(ctx, createJob,
+		arg.ID,
+		arg.Kind,
+		arg.Payload,
+		arg.CreatedAt,
+		arg.UpdatedAt,
+	)
+	return err
+}
+
+const failJob = `-- name: FailJob :execrows
+UPDATE jobs
+SET status = 'failed', error = ?1, finished_at = ?2, updated_at = ?2
+WHERE id = ?3 AND status = 'running'
+`
+
+type FailJobParams struct {
+	Error sql.NullString
+	Now   sql.NullString
+	ID    string
+}
+
+func (q *Queries) FailJob(ctx context.Context, arg FailJobParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, failJob, arg.Error, arg.Now, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const failRunningJobs = `-- name: FailRunningJobs :execrows
+UPDATE jobs
+SET status = 'failed', error = ?1, finished_at = ?2, updated_at = ?2
+WHERE status = 'running'
+`
+
+type FailRunningJobsParams struct {
+	Error sql.NullString
+	Now   sql.NullString
+}
+
+// Fails every job left running, at boot before any worker starts.
+func (q *Queries) FailRunningJobs(ctx context.Context, arg FailRunningJobsParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, failRunningJobs, arg.Error, arg.Now)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const getJob = `-- name: GetJob :one
+SELECT id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at FROM jobs
+WHERE id = ?
+`
+
+func (q *Queries) GetJob(ctx context.Context, id string) (Job, error) {
+	row := q.db.QueryRowContext(ctx, getJob, id)
+	var i Job
+	err := row.Scan(
+		&i.ID,
+		&i.Kind,
+		&i.Status,
+		&i.Payload,
+		&i.Result,
+		&i.Error,
+		&i.StartedAt,
+		&i.FinishedAt,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
+const publishJobResult = `-- name: PublishJobResult :execrows
+UPDATE jobs
+SET result = ?, updated_at = ?
+WHERE id = ? AND status = 'running'
+`
+
+type PublishJobResultParams struct {
+	Result    sql.NullString
+	UpdatedAt string
+	ID        string
+}
+
+// Stores a partial result of a running job.
+func (q *Queries) PublishJobResult(ctx context.Context, arg PublishJobResultParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, publishJobResult, arg.Result, arg.UpdatedAt, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const succeedJob = `-- name: SucceedJob :execrows
+UPDATE jobs
+SET status = 'succeeded', result = ?1, finished_at = ?2, updated_at = ?2
+WHERE id = ?3 AND status = 'running'
+`
+
+type SucceedJobParams struct {
+	Result sql.NullString
+	Now    sql.NullString
+	ID     string
+}
+
+func (q *Queries) SucceedJob(ctx context.Context, arg SucceedJobParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, succeedJob, arg.Result, arg.Now, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}

internal/db/querier.go

@@ -6,17 +6,25 @@ package db
 
 import (
 	"context"
+	"database/sql"
 )
 
 type Querier interface {
 	// An unknown id returns sql.ErrNoRows.
 	ArchiveDinner(ctx context.Context, arg ArchiveDinnerParams) (Dinner, error)
+	// Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
+	// when no job is queued.
+	ClaimNextJob(ctx context.Context, now sql.NullString) (Job, error)
 	CreateDinner(ctx context.Context, arg CreateDinnerParams) (Dinner, error)
+	CreateJob(ctx context.Context, arg CreateJobParams) error
 	CreatePhoto(ctx context.Context, arg CreatePhotoParams) (Photo, error)
 	CreateRecipeIngredient(ctx context.Context, arg CreateRecipeIngredientParams) error
 	CreateRecipeStep(ctx context.Context, arg CreateRecipeStepParams) error
 	DeleteRecipeIngredients(ctx context.Context, recipeID string) error
 	DeleteRecipeSteps(ctx context.Context, recipeID string) error
+	FailJob(ctx context.Context, arg FailJobParams) (int64, error)
+	// Fails every job left running, at boot before any worker starts.
+	FailRunningJobs(ctx context.Context, arg FailRunningJobsParams) (int64, error)
 	// Returns sql.ErrNoRows until the row exists.
 	GetAISettings(ctx context.Context) (AiSetting, error)
 	// Returns sql.ErrNoRows until the row exists.
@@ -26,6 +34,7 @@ type Querier interface {
 	GetHouseholdSettings(ctx context.Context) (HouseholdSetting, error)
 	// Returns sql.ErrNoRows until the row exists.
 	GetIntegrationsGoogle(ctx context.Context) (IntegrationsGoogle, error)
+	GetJob(ctx context.Context, id string) (Job, error)
 	// Finds stored image bytes by content hash. The schema does not make sha256
 	// unique, so the earliest-created (lowest id) match is returned.
 	GetPhotoBySHA256(ctx context.Context, sha256 string) (Photo, error)
@@ -36,8 +45,11 @@ type Querier interface {
 	ListDinnersByStatus(ctx context.Context, status string) ([]Dinner, error)
 	ListRecipeIngredients(ctx context.Context, recipeID string) ([]RecipeIngredient, error)
 	ListRecipeSteps(ctx context.Context, recipeID string) ([]RecipeStep, error)
+	// Stores a partial result of a running job.
+	PublishJobResult(ctx context.Context, arg PublishJobResultParams) (int64, error)
 	// An unknown id returns sql.ErrNoRows.
 	SetDinnerFavorite(ctx context.Context, arg SetDinnerFavoriteParams) (Dinner, error)
+	SucceedJob(ctx context.Context, arg SucceedJobParams) (int64, error)
 	// An unknown id returns sql.ErrNoRows.
 	UpdateDinner(ctx context.Context, arg UpdateDinnerParams) (Dinner, error)
 	// Creates the single row (id 1) or replaces every column except id and

internal/jobs/builtin.go

@@ -0,0 +1,44 @@
+package jobs
+
+import (
+	"context"
+	"encoding/json"
+)
+
+// KindNoop is the framework's built-in job kind: it does no work and
+// succeeds with {"ok":true}. It is the runner's own trivial kind, as Feature
+// 09's trivial test skill is the harness's, and the startup and readiness
+// job: a noop job reaching succeeded proves the runner claims, runs and
+// records jobs.
+const KindNoop = "noop"
+
+// NoopPayload is the payload of a noop job.
+type NoopPayload struct{}
+
+// NoopResult is the result of a noop job.
+type NoopResult struct{}
+
+// PublicJSON returns {"ok":true}.
+func (NoopResult) PublicJSON() ([]byte, error) {
+	return json.Marshal(struct {
+		OK bool `json:"ok"`
+	}{OK: true})
+}
+
+func runNoop(context.Context, NoopPayload, func(NoopResult) error) (NoopResult, error) {
+	return NoopResult{}, nil
+}
+
+// Builtins enqueues the framework's built-in job kinds.
+type Builtins struct {
+	Noop *Kind[NoopPayload]
+}
+
+// RegisterBuiltins registers the framework's built-in job kinds on r.
+func RegisterBuiltins(r *Runner) (*Builtins, error) {
+	noop, err := Register(r, KindNoop, runNoop)
+	if err != nil {
+		return nil, err
+	}
+	return &Builtins{Noop: noop}, nil
+}

internal/jobs/jobs.go

@@ -0,0 +1,461 @@
+// Package jobs runs long work in the background. A job is a row in the jobs
+// table: Enqueue inserts it queued, and a fixed pool of workers claims queued
+// jobs oldest first, so the table itself is the queue and a job enqueued
+// before a restart still runs after it.
+//
+// Each job kind has one handler, registered before Start. A handler takes a
+// typed payload and returns a result implementing Public; it may publish
+// partial results while it runs. The runner stores only what a result's
+// PublicJSON returns and, for a failed job, only a failure code: either a
+// *Failure code the kind declared at registration or one of the runner's own
+// codes. Handler error text, which may quote a URL, token or response body
+// from a dependency, is never stored, returned by the API or logged.
+package jobs
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"regexp"
+	"sync"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+)
+
+// Status is the state of a job.
+type Status string
+
+// Job states. A job moves queued → running → succeeded or failed.
+const (
+	StatusQueued    Status = "queued"
+	StatusRunning   Status = "running"
+	StatusSucceeded Status = "succeeded"
+	StatusFailed    Status = "failed"
+)
+
+// Failure codes the runner stores itself. A kind cannot declare them.
+const (
+	// CodeInterrupted is the error of a job a previous process left running.
+	// It is fixed framework text, not a snake-case code, so no kind can
+	// declare it.
+	CodeInterrupted = "interrupted by restart"
+	// CodeInternal is the error of a job whose handler failed with anything
+	// but a *Failure with a code its kind declared, or whose payload or
+	// result could not be encoded. The underlying error is neither stored
+	// nor logged.
+	CodeInternal = "internal_error"
+	// CodeUnknownKind is the error of a job whose kind has no handler.
+	CodeUnknownKind = "unknown_kind"
+)
+
+// Claim retry backoff. A worker whose claim fails with a database error
+// retries after claimRetryMin, doubling up to claimRetryMax, so a job queued
+// when the error clears is claimed without waiting for a wake token.
+const (
+	claimRetryMin = 50 * time.Millisecond
+	claimRetryMax = 5 * time.Second
+)
+
+// codePattern is the form of every failure code: lower-case snake case.
+var codePattern = regexp.MustCompile(`^[a-z][a-z0-9_]{0,63}$`)
+
+// ErrNotFound is returned by Get for an unknown job id.
+var ErrNotFound = errors.New("jobs: job not found")
+
+// ErrUnknownKind is returned by Runner.Enqueue for a kind with no
+// registered handler.
+var ErrUnknownKind = errors.New("jobs: unknown job kind")
+
+// ErrInvalidPayload is returned by Runner.Enqueue for a payload that does
+// not decode into the kind's payload type.
+var ErrInvalidPayload = errors.New("jobs: payload does not decode")
+
+// Failure is a handler error reporting a failure code the household may
+// see. Code must be one the job's kind declared at registration; the runner
+// stores it as the job's error. Any other handler error, including a Failure
+// with an undeclared code, is stored as CodeInternal.
+type Failure struct {
+	Code string
+}
+
+func (f *Failure) Error() string { return "jobs: failure " + f.Code }
+
+// Public is a job result. PublicJSON returns the JSON the runner stores and
+// GET /api/v1/jobs/{id} returns, so a result type states exactly which of its
+// data is public; nothing else of the value is ever encoded.
+type Public interface {
+	PublicJSON() ([]byte, error)
+}
+
+// Handler runs one job with its payload. It may call publish any number of
+// times with a partial result, which Get returns while the job runs; its
+// return value is the final result. ctx is cancelled when the runner stops.
+type Handler[P any, R Public] func(ctx context.Context, payload P, publish func(partial R) error) (R, error)
+
+// Job is a job as reported by GET /api/v1/jobs/{id}. Result is the latest
+// partial or final result, and is null until the handler publishes one.
+// Error is set only on a failed job.
+type Job struct {
+	ID     string          `json:"id"`
+	Kind   string          `json:"kind"`
+	Status Status          `json:"status"`
+	Result json.RawMessage `json:"result"`
+	Error  *string         `json:"error"`
+}
+
+// runFunc is a registered Handler with its payload and results as JSON. It
+// returns the failure code to store, or "" on success; the logged reason
+// never quotes an error.
+type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (result json.RawMessage, code, reason string)
+
+// handler is a registered kind: run runs a job, and check reports whether
+// a payload decodes into the kind's payload type.
+type handler struct {
+	run   runFunc
+	check func(payload []byte) bool
+}
+
+// Runner owns the job handlers and the worker pool.
+type Runner struct {
+	st      *store.Store
+	logger  *slog.Logger
+	workers int
+	// wake holds up to one token per worker; Enqueue adds one so an idle
+	// worker looks for the new job.
+	wake chan struct{}
+	// claim claims the oldest queued job, marking it running at now.
+	claim func(ctx context.Context, now sql.NullString) (db.Job, error)
+
+	mu       sync.Mutex
+	handlers map[string]handler
+	started  bool
+
+	wg sync.WaitGroup
+}
+
+// NewRunner returns a runner over st with workers concurrent workers.
+func NewRunner(st *store.Store, logger *slog.Logger, workers int) (*Runner, error) {
+	if st == nil {
+		return nil, errors.New("jobs: nil store")
+	}
+	if logger == nil {
+		return nil, errors.New("jobs: nil logger")
+	}
+	if workers < 1 {
+		return nil, fmt.Errorf("jobs: %d workers, want at least 1", workers)
+	}
+	return &Runner{
+		st:       st,
+		logger:   logger,
+		workers:  workers,
+		wake:     make(chan struct{}, workers),
+		claim:    db.New(st.Writer).ClaimNextJob,
+		handlers: make(map[string]handler),
+	}, nil
+}
+
+// Kind enqueues jobs of one registered kind.
+type Kind[P any] struct {
+	r    *Runner
+	name string
+}
+
+// Register makes h the handler for kind and returns the Kind that enqueues
+// its jobs. failureCodes are the *Failure codes h may report. It must be
+// called before Start, once per kind.
+func Register[P any, R Public](r *Runner, kind string, h Handler[P, R], failureCodes ...string) (*Kind[P], error) {
+	if kind == "" {
+		return nil, errors.New("jobs: register: empty kind")
+	}
+	if h == nil {
+		return nil, fmt.Errorf("jobs: register %q: nil handler", kind)
+	}
+	declared := make(map[string]bool, len(failureCodes))
+	for _, code := range failureCodes {
+		if !codePattern.MatchString(code) {
+			return nil, fmt.Errorf("jobs: register %q: failure code %q is not lower-case snake case", kind, code)
+		}
+		if code == CodeInternal || code == CodeUnknownKind {
+			return nil, fmt.Errorf("jobs: register %q: failure code %q is reserved for the runner", kind, code)
+		}
+		declared[code] = true
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.started {
+		return nil, fmt.Errorf("jobs: register %q: runner already started", kind)
+	}
+	if _, ok := r.handlers[kind]; ok {
+		return nil, fmt.Errorf("jobs: register %q: kind already registered", kind)
+	}
+	run := func(ctx context.Context, payload string, publish func(json.RawMessage) error) (json.RawMessage, string, string) {
+		p, err := decodePayload[P]([]byte(payload))
+		if err != nil {
+			return nil, CodeInternal, "payload does not decode"
+		}
+		result, err := h(ctx, p, func(partial R) error {
+			data, err := publicJSON(partial)
+			if err != nil {
+				return err
+			}
+			return publish(data)
+		})
+		if err != nil {
+			var failure *Failure
+			if !errors.As(err, &failure) {
+				return nil, CodeInternal, "handler error"
+			}
+			if !declared[failure.Code] {
+				return nil, CodeInternal, "undeclared failure code"
+			}
+			return nil, failure.Code, "handler failure"
+		}
+		data, err := publicJSON(result)
+		if err != nil {
+			return nil, CodeInternal, "result does not encode"
+		}
+		return data, "", ""
+	}
+	check := func(payload []byte) bool {
+		_, err := decodePayload[P](payload)
+		return err == nil
+	}
+	r.handlers[kind] = handler{run: run, check: check}
+	return &Kind[P]{r: r, name: kind}, nil
+}
+
+// decodePayload decodes payload into P, refusing unknown fields.
+func decodePayload[P any](payload []byte) (P, error) {
+	var p P
+	dec := json.NewDecoder(bytes.NewReader(payload))
+	dec.DisallowUnknownFields()
+	err := dec.Decode(&p)
+	return p, err
+}
+
+// publicJSON returns v's PublicJSON, refusing anything that is not one JSON
+// value. Its error never quotes PublicJSON's error.
+func publicJSON(v Public) (json.RawMessage, error) {
+	data, err := v.PublicJSON()
+	if err != nil || !json.Valid(data) {
+		return nil, errors.New("jobs: result has no valid public JSON")
+	}
+	return data, nil
+}
+
+// Enqueue stores a queued job with payload and returns its id. A job
+// enqueued before Start waits for it.
+func (k *Kind[P]) Enqueue(ctx context.Context, payload P) (string, error) {
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return "", fmt.Errorf("jobs: enqueue %s: encode payload: %w", k.name, err)
+	}
+	return k.r.enqueue(ctx, k.name, data)
+}
+
+// Enqueue stores a queued job of kind with its JSON payload and returns its
+// id, as Kind.Enqueue does for a kind known at compile time. It returns
+// ErrUnknownKind for a kind with no registered handler and
+// ErrInvalidPayload for a payload that does not decode into the kind's
+// payload type; neither stores a job.
+func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessage) (string, error) {
+	r.mu.Lock()
+	h, ok := r.handlers[kind]
+	r.mu.Unlock()
+	if !ok {
+		return "", ErrUnknownKind
+	}
+	if !h.check(payload) {
+		return "", ErrInvalidPayload
+	}
+	return r.enqueue(ctx, kind, payload)
+}
+
+// enqueue stores a queued job of kind with its encoded payload and wakes a
+// worker.
+func (r *Runner) enqueue(ctx context.Context, kind string, payload []byte) (string, error) {
+	id := store.NewID()
+	now := store.Now().String()
+	if err := db.New(r.st.Writer).CreateJob(ctx, db.CreateJobParams{
+		ID:        id,
+		Kind:      kind,
+		Payload:   string(payload),
+		CreatedAt: now,
+		UpdatedAt: now,
+	}); err != nil {
+		return "", fmt.Errorf("jobs: enqueue %s: %w", kind, err)
+	}
+	select {
+	case r.wake <- struct{}{}:
+	default: // every worker already has a token and will look again
+	}
+	return id, nil
+}
+
+// Start fails every job a previous process left running with
+// CodeInterrupted, then starts the workers. They stop when ctx is cancelled;
+// Wait waits for them. A job whose handler returns an error after ctx is
+// cancelled is left running, so the next Start fails it as interrupted.
+func (r *Runner) Start(ctx context.Context) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.started {
+		return errors.New("jobs: runner already started")
+	}
+	now := store.Now().String()
+	n, err := db.New(r.st.Writer).FailRunningJobs(ctx, db.FailRunningJobsParams{
+		Error: sql.NullString{String: CodeInterrupted, Valid: true},
+		Now:   sql.NullString{String: now, Valid: true},
+	})
+	if err != nil {
+		return fmt.Errorf("jobs: fail interrupted jobs: %w", err)
+	}
+	if n > 0 {
+		r.logger.WarnContext(ctx, "failed jobs interrupted by restart", "count", n)
+	}
+	r.started = true
+	for range r.workers {
+		r.wg.Add(1)
+		go r.work(ctx)
+	}
+	return nil
+}
+
+// Wait blocks until every worker has stopped.
+func (r *Runner) Wait() {
+	r.wg.Wait()
+}
+
+// Get returns the job with id, or ErrNotFound.
+func (r *Runner) Get(ctx context.Context, id string) (Job, error) {
+	row, err := db.New(r.st.Reader).GetJob(ctx, id)
+	if errors.Is(err, sql.ErrNoRows) {
+		return Job{}, ErrNotFound
+	}
+	if err != nil {
+		return Job{}, fmt.Errorf("jobs: read job: %w", err)
+	}
+	job := Job{ID: row.ID, Kind: row.Kind, Status: Status(row.Status)}
+	if row.Result.Valid {
+		job.Result = json.RawMessage(row.Result.String)
+	}
+	if row.Error.Valid {
+		job.Error = &row.Error.String
+	}
+	return job, nil
+}
+
+// work claims and runs queued jobs until ctx is cancelled, waiting for a
+// wake token whenever none is queued. A claim that fails with a database
+// error is retried with backoff rather than waiting for a token, so a job
+// already queued is not stranded by a transient error.
+func (r *Runner) work(ctx context.Context) {
+	defer r.wg.Done()
+	backoff := claimRetryMin
+	for ctx.Err() == nil {
+		job, err := r.claim(ctx, sql.NullString{String: store.Now().String(), Valid: true})
+		switch {
+		case err == nil:
+			backoff = claimRetryMin
+			r.run(ctx, job)
+			continue
+		case errors.Is(err, sql.ErrNoRows):
+			backoff = claimRetryMin
+			select {
+			case <-r.wake:
+			case <-ctx.Done():
+			}
+			continue
+		case ctx.Err() != nil:
+			return
+		}
+		r.logger.ErrorContext(ctx, "claim job failed; retrying", "reason", "database error", "retry_in", backoff)
+		timer := time.NewTimer(backoff)
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			timer.Stop()
+		}
+		backoff = min(2*backoff, claimRetryMax)
+	}
+}
+
+// run runs a claimed job to completion. The final write is made even when
+// ctx has been cancelled, so a handler that finishes during shutdown is
+// recorded.
+func (r *Runner) run(ctx context.Context, job db.Job) {
+	write := context.WithoutCancel(ctx)
+	logger := r.logger.With("job_id", job.ID, "kind", job.Kind)
+	q := db.New(r.st.Writer)
+
+	h, ok := r.handlers[job.Kind] // read-only once started
+	if !ok {
+		logger.ErrorContext(ctx, "job failed", "code", CodeUnknownKind, "reason", "no handler registered for the kind")
+		r.fail(write, logger, job.ID, CodeUnknownKind)
+		return
+	}
+
+	publish := func(partial json.RawMessage) error {
+		n, err := q.PublishJobResult(ctx, db.PublishJobResultParams{
+			Result:    sql.NullString{String: string(partial), Valid: true},
+			UpdatedAt: store.Now().String(),
+			ID:        job.ID,
+		})
+		if err != nil {
+			return fmt.Errorf("jobs: publish result: %w", err)
+		}
+		if n != 1 {
+			return fmt.Errorf("jobs: publish result: job %s is not running", job.ID)
+		}
+		return nil
+	}
+
+	result, code, reason := h.run(ctx, job.Payload, publish)
+	if code != "" {
+		if ctx.Err() != nil {
+			logger.WarnContext(write, "job interrupted by shutdown; it fails at the next start")
+			return
+		}
+		level := slog.LevelInfo
+		if code == CodeInternal {
+			level = slog.LevelError
+		}
+		logger.Log(ctx, level, "job failed", "code", code, "reason", reason)
+		r.fail(write, logger, job.ID, code)
+		return
+	}
+
+	n, err := q.SucceedJob(write, db.SucceedJobParams{
+		Result: sql.NullString{String: string(result), Valid: true},
+		Now:    sql.NullString{String: store.Now().String(), Valid: true},
+		ID:     job.ID,
+	})
+	if err == nil && n != 1 {
+		err = errors.New("job is not running")
+	}
+	if err != nil {
+		logger.ErrorContext(write, "record job success", "error", err)
+	}
+}
+
+// fail records the job as failed with code. The record errors it logs come
+// from the database, never from a handler.
+func (r *Runner) fail(ctx context.Context, logger *slog.Logger, id, code string) {
+	n, err := db.New(r.st.Writer).FailJob(ctx, db.FailJobParams{
+		Error: sql.NullString{String: code, Valid: true},
+		Now:   sql.NullString{String: store.Now().String(), Valid: true},
+		ID:    id,
+	})
+	if err == nil && n != 1 {
+		err = errors.New("job is not running")
+	}
+	if err != nil {
+		logger.ErrorContext(ctx, "record job failure", "error", err)
+	}
+}

internal/jobs/jobs_test.go

@@ -0,0 +1,606 @@
+package jobs
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
+)
+
+// sentinelSecret stands in for a credential a handler holds while it runs.
+const sentinelSecret = "sk-sentinel-5f2c9a"
+
+type countPayload struct {
+	Steps int `json:"steps"`
+}
+
+type countResult struct {
+	Done int `json:"done"`
+}
+
+func (c countResult) PublicJSON() ([]byte, error) { return json.Marshal(c) }
+
+func migratedStore(t *testing.T) *store.Store {
+	t.Helper()
+	ctx := context.Background()
+	st, err := store.Open(ctx, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := st.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	if err := store.Migrate(ctx, st.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	return st
+}
+
+func newRunner(t *testing.T, st *store.Store, workers int) *Runner {
+	t.Helper()
+	return newRunnerLogging(t, st, workers, io.Discard)
+}
+
+func newRunnerLogging(t *testing.T, st *store.Store, workers int, w io.Writer) *Runner {
+	t.Helper()
+	r, err := NewRunner(st, slog.New(slog.NewTextHandler(w, nil)), workers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
+// logBuffer collects log output written by concurrent workers.
+type logBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *logBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *logBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+// start starts r and stops it when the test ends. The returned function
+// stops it early and waits for its workers.
+func start(t *testing.T, r *Runner) (stop func()) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	if err := r.Start(ctx); err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	stop = func() {
+		cancel()
+		r.Wait()
+	}
+	t.Cleanup(stop)
+	return stop
+}
+
+func register[P any, R Public](t *testing.T, r *Runner, kind string, h Handler[P, R], failureCodes ...string) *Kind[P] {
+	t.Helper()
+	k, err := Register(r, kind, h, failureCodes...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return k
+}
+
+func enqueue[P any](t *testing.T, k *Kind[P], payload P) string {
+	t.Helper()
+	id, err := k.Enqueue(context.Background(), payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return id
+}
+
+// waitFor polls Get until done accepts the job, failing after five seconds.
+func waitFor(t *testing.T, r *Runner, id string, done func(Job) bool) Job {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		job, err := r.Get(context.Background(), id)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if done(job) {
+			return job
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("job %s never reached the expected state; last %+v (result %s)", id, job, job.Result)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func finished(j Job) bool { return j.Status == StatusSucceeded || j.Status == StatusFailed }
+
+// rawRow returns every column of the job's row joined, for leak checks.
+func rawRow(t *testing.T, st *store.Store, id string) string {
+	t.Helper()
+	var cols [10]any
+	ptrs := make([]any, len(cols))
+	for i := range cols {
+		ptrs[i] = &cols[i]
+	}
+	if err := st.Reader.QueryRow("SELECT * FROM jobs WHERE id = ?", id).Scan(ptrs...); err != nil {
+		t.Fatal(err)
+	}
+	return fmt.Sprint(cols[:]...)
+}
+
+func TestJobSucceeds(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	r := newRunner(t, st, 2)
+	count := register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
+		return countResult{Done: p.Steps}, nil
+	})
+	start(t, r)
+
+	id := enqueue(t, count, countPayload{Steps: 3})
+	job := waitFor(t, r, id, finished)
+	if job.ID != id || job.Kind != "count" || job.Status != StatusSucceeded || job.Error != nil {
+		t.Fatalf("job = %+v, want succeeded count job %s with no error", job, id)
+	}
+	if string(job.Result) != `{"done":3}` {
+		t.Errorf("result = %s, want {\"done\":3}", job.Result)
+	}
+
+	var startedAt, finishedAt, createdAt, updatedAt, payload string
+	if err := st.Reader.QueryRow("SELECT started_at, finished_at, created_at, updated_at, payload FROM jobs WHERE id = ?", id).
+		Scan(&startedAt, &finishedAt, &createdAt, &updatedAt, &payload); err != nil {
+		t.Fatal(err)
+	}
+	for name, v := range map[string]string{"started_at": startedAt, "finished_at": finishedAt, "created_at": createdAt, "updated_at": updatedAt} {
+		var ts store.Timestamp
+		if err := ts.Scan(v); err != nil {
+			t.Errorf("%s = %q: %v", name, v, err)
+		}
+	}
+	if payload != `{"steps":3}` {
+		t.Errorf("payload = %s, want {\"steps\":3}", payload)
+	}
+}
+
+func TestJobFailsWithFailureCode(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	r := newRunner(t, st, 1)
+	k := register(t, r, "suggest", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, fmt.Errorf("suggest: %w", &Failure{Code: "no_regulars"})
+	}, "no_regulars")
+	start(t, r)
+
+	job := waitFor(t, r, enqueue(t, k, countPayload{}), finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != "no_regulars" {
+		t.Fatalf("job = %+v, want failed with code no_regulars", job)
+	}
+	if job.Result != nil {
+		t.Errorf("result = %s, want none", job.Result)
+	}
+	var stored string
+	if err := st.Reader.QueryRow("SELECT error FROM jobs WHERE id = ?", job.ID).Scan(&stored); err != nil {
+		t.Fatal(err)
+	}
+	if stored != "no_regulars" {
+		t.Errorf("stored error = %q, want only the code", stored)
+	}
+}
+
+func TestUndeclaredFailureCodeIsInternal(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	k := register(t, r, "suggest", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, &Failure{Code: "not_declared"}
+	}, "no_regulars")
+	start(t, r)
+
+	job := waitFor(t, r, enqueue(t, k, countPayload{}), finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != CodeInternal {
+		t.Fatalf("job = %+v, want failed with %q", job, CodeInternal)
+	}
+}
+
+// credential holds a secret the way config.Secret does: unexported, so
+// encoding/json cannot see it.
+type credential struct{ value string }
+
+func (credential) String() string { return "[REDACTED]" }
+
+// leakyResult holds a secret in exported and unexported fields; only
+// Dinners is public.
+type leakyResult struct {
+	Dinners []string
+	Key     credential
+	Token   string
+	apiKey  string
+}
+
+func (l leakyResult) PublicJSON() ([]byte, error) {
+	return json.Marshal(struct {
+		Dinners []string `json:"dinners"`
+	}{l.Dinners})
+}
+
+// badResult has no valid public JSON; its encoding error quotes a secret.
+type badResult struct{}
+
+func (badResult) PublicJSON() ([]byte, error) {
+	return nil, fmt.Errorf("encode with key %s", sentinelSecret)
+}
+
+// TestSecretsNeverStoredOrLogged proves a handler holding a secret cannot
+// surface it in the jobs row, Get's result or the log through its result,
+// partial results, *Failure, error text or result encoding error.
+func TestSecretsNeverStoredOrLogged(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	var logs logBuffer
+	r := newRunnerLogging(t, st, 1, &logs)
+	secretResult := leakyResult{Dinners: []string{"Tacos"}, Key: credential{sentinelSecret}, Token: sentinelSecret, apiKey: sentinelSecret}
+	typed := register(t, r, "typed", func(_ context.Context, _ countPayload, publish func(leakyResult) error) (leakyResult, error) {
+		if err := publish(secretResult); err != nil {
+			return leakyResult{}, err
+		}
+		return secretResult, nil
+	})
+	leakyError := register(t, r, "leaky-error", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) {
+		return leakyResult{}, fmt.Errorf("GET https://api.example.com/v1?key=%s: 401 %s", sentinelSecret, sentinelSecret)
+	})
+	leakyFailure := register(t, r, "leaky-failure", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) {
+		return leakyResult{}, &Failure{Code: sentinelSecret}
+	}, "quota_exceeded")
+	failure := register(t, r, "failure", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) {
+		return leakyResult{}, fmt.Errorf("call with key %s: %w", sentinelSecret, &Failure{Code: "quota_exceeded"})
+	}, "quota_exceeded")
+	badEncoding := register(t, r, "bad-encoding", func(context.Context, countPayload, func(badResult) error) (badResult, error) {
+		return badResult{}, nil
+	})
+	stop := start(t, r)
+
+	cases := []struct {
+		kind       *Kind[countPayload]
+		wantStatus Status
+		wantResult string
+		wantError  string
+	}{
+		{typed, StatusSucceeded, `{"dinners":["Tacos"]}`, ""},
+		{leakyError, StatusFailed, "", CodeInternal},
+		{leakyFailure, StatusFailed, "", CodeInternal},
+		{failure, StatusFailed, "", "quota_exceeded"},
+		{badEncoding, StatusFailed, "", CodeInternal},
+	}
+	var ids []string
+	for _, c := range cases {
+		job := waitFor(t, r, enqueue(t, c.kind, countPayload{}), finished)
+		ids = append(ids, job.ID)
+		gotError := ""
+		if job.Error != nil {
+			gotError = *job.Error
+		}
+		if job.Status != c.wantStatus || string(job.Result) != c.wantResult || gotError != c.wantError {
+			t.Errorf("%s job = %+v (result %s), want %s with result %q and error %q", job.Kind, job, job.Result, c.wantStatus, c.wantResult, c.wantError)
+		}
+		body, err := json.Marshal(job)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if strings.Contains(string(body), sentinelSecret) {
+			t.Errorf("%s job as served contains the secret: %s", job.Kind, body)
+		}
+	}
+	stop()
+	for _, id := range ids {
+		if row := rawRow(t, st, id); strings.Contains(row, sentinelSecret) {
+			t.Errorf("job %s row contains the secret: %s", id, row)
+		}
+	}
+	if strings.Contains(logs.String(), sentinelSecret) {
+		t.Errorf("log contains the secret:\n%s", logs.String())
+	}
+	if !strings.Contains(logs.String(), "code="+CodeInternal) {
+		t.Errorf("log does not record the internal_error code:\n%s", logs.String())
+	}
+}
+
+func TestProgressiveResultsVisibleMidRun(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	release := make(chan struct{})
+	k := register(t, r, "draft", func(ctx context.Context, p countPayload, publish func(countResult) error) (countResult, error) {
+		for i := 1; i < p.Steps; i++ {
+			if err := publish(countResult{Done: i}); err != nil {
+				return countResult{}, err
+			}
+			select {
+			case <-release:
+			case <-ctx.Done():
+				return countResult{}, ctx.Err()
+			}
+		}
+		return countResult{Done: p.Steps}, nil
+	})
+	start(t, r)
+
+	id := enqueue(t, k, countPayload{Steps: 3})
+	for step := 1; step < 3; step++ {
+		want := fmt.Sprintf(`{"done":%d}`, step)
+		job := waitFor(t, r, id, func(j Job) bool { return string(j.Result) == want })
+		if job.Status != StatusRunning || job.Error != nil {
+			t.Fatalf("mid-run job = %+v, want running with no error", job)
+		}
+		release <- struct{}{}
+	}
+	job := waitFor(t, r, id, finished)
+	if job.Status != StatusSucceeded || string(job.Result) != `{"done":3}` {
+		t.Fatalf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result)
+	}
+}
+
+func TestRestartFailsInterruptedJobsAndRunsQueuedOnes(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+
+	first := newRunner(t, st, 1)
+	started := make(chan struct{})
+	k := register(t, first, "slow", func(ctx context.Context, _ countPayload, publish func(countResult) error) (countResult, error) {
+		if err := publish(countResult{Done: 1}); err != nil {
+			return countResult{}, err
+		}
+		close(started)
+		<-ctx.Done()
+		return countResult{}, ctx.Err()
+	})
+	stop := start(t, first)
+	interrupted := enqueue(t, k, countPayload{})
+	<-started
+	waiting := enqueue(t, k, countPayload{}) // the only worker is busy
+	stop()
+
+	if job := waitFor(t, first, interrupted, func(Job) bool { return true }); job.Status != StatusRunning {
+		t.Fatalf("after shutdown job = %+v, want left running", job)
+	}
+
+	second := newRunner(t, st, 1)
+	register(t, second, "slow", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{Done: 9}, nil
+	})
+	start(t, second)
+
+	job := waitFor(t, second, interrupted, finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != "interrupted by restart" {
+		t.Errorf("interrupted job = %+v, want failed with %q", job, "interrupted by restart")
+	}
+	if string(job.Result) != `{"done":1}` {
+		t.Errorf("interrupted job result = %s, want the last partial result kept", job.Result)
+	}
+	if job := waitFor(t, second, waiting, finished); job.Status != StatusSucceeded {
+		t.Errorf("queued job = %+v, want run by the new runner", job)
+	}
+}
+
+func TestWorkerPoolIsBounded(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 2)
+	var running, peak atomic.Int32
+	release := make(chan struct{})
+	k := register(t, r, "busy", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		n := running.Add(1)
+		for {
+			p := peak.Load()
+			if n <= p || peak.CompareAndSwap(p, n) {
+				break
+			}
+		}
+		<-release
+		running.Add(-1)
+		return countResult{}, nil
+	})
+	start(t, r)
+
+	ids := make([]string, 5)
+	for i := range ids {
+		ids[i] = enqueue(t, k, countPayload{})
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for running.Load() < 2 {
+		if time.Now().After(deadline) {
+			t.Fatal("two workers never ran at once")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	close(release)
+	for _, id := range ids {
+		if job := waitFor(t, r, id, finished); job.Status != StatusSucceeded {
+			t.Errorf("job = %+v, want succeeded", job)
+		}
+	}
+	if got := peak.Load(); got != 2 {
+		t.Errorf("peak concurrent jobs = %d, want 2", got)
+	}
+}
+
+func TestJobWithoutHandlerFails(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	old := newRunner(t, st, 1)
+	k := register(t, old, "retired", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, nil
+	})
+	id := enqueue(t, k, countPayload{}) // never started
+
+	r := newRunner(t, st, 1)
+	start(t, r)
+	job := waitFor(t, r, id, finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != CodeUnknownKind {
+		t.Errorf("job = %+v, want failed for a missing handler", job)
+	}
+}
+
+func TestGetUnknownID(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	if _, err := r.Get(context.Background(), store.NewID()); !errors.Is(err, ErrNotFound) {
+		t.Errorf("Get unknown id error = %v, want ErrNotFound", err)
+	}
+}
+
+func TestRegisterRefusesBadRegistrations(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	h := func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, nil
+	}
+	register(t, r, "count", h)
+
+	if _, err := Register(r, "count", h); err == nil {
+		t.Error("duplicate kind registered")
+	}
+	if _, err := Register(r, "", h); err == nil {
+		t.Error("empty kind registered")
+	}
+	if _, err := Register[countPayload, countResult](r, "nil", nil); err == nil {
+		t.Error("nil handler registered")
+	}
+	for _, code := range []string{"", "Bad Code", "sk-sentinel", CodeInternal, CodeInterrupted, CodeUnknownKind} {
+		if _, err := Register(r, "code", h, code); err == nil {
+			t.Errorf("failure code %q accepted", code)
+		}
+	}
+	start(t, r)
+	if _, err := Register(r, "late", h); err == nil {
+		t.Error("kind registered after Start")
+	}
+	if err := r.Start(context.Background()); err == nil {
+		t.Error("second Start succeeded")
+	}
+}
+
+func TestNewRunnerRefusesBadConfig(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	if _, err := NewRunner(st, logger, 0); err == nil {
+		t.Error("zero workers accepted")
+	}
+	if _, err := NewRunner(nil, logger, 1); err == nil {
+		t.Error("nil store accepted")
+	}
+	if _, err := NewRunner(st, nil, 1); err == nil {
+		t.Error("nil logger accepted")
+	}
+}
+
+func TestNoopSucceeds(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	builtins, err := RegisterBuiltins(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	start(t, r)
+
+	job := waitFor(t, r, enqueue(t, builtins.Noop, NoopPayload{}), finished)
+	if job.Kind != KindNoop || job.Status != StatusSucceeded || string(job.Result) != `{"ok":true}` || job.Error != nil {
+		t.Errorf("noop job = %+v (result %s), want succeeded with {\"ok\":true}", job, job.Result)
+	}
+}
+
+func TestClaimErrorIsRetried(t *testing.T) {
+	st := migratedStore(t)
+	logs := &logBuffer{}
+	r := newRunnerLogging(t, st, 1, logs)
+	k := register(t, r, "count", func(ctx context.Context, p countPayload, publish func(countResult) error) (countResult, error) {
+		return countResult{Done: p.Steps}, nil
+	})
+	id := enqueue(t, k, countPayload{Steps: 3})
+	// Drop the wake token, as for a job queued before a restart, so only
+	// the retry can find the job.
+	<-r.wake
+
+	var claims atomic.Int32
+	claim := r.claim
+	r.claim = func(ctx context.Context, now sql.NullString) (db.Job, error) {
+		if claims.Add(1) == 1 {
+			return db.Job{}, errors.New("database is locked " + sentinelSecret)
+		}
+		return claim(ctx, now)
+	}
+	start(t, r)
+
+	job := waitFor(t, r, id, finished)
+	if job.Status != StatusSucceeded || string(job.Result) != `{"done":3}` {
+		t.Fatalf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result)
+	}
+	if n := claims.Load(); n < 2 {
+		t.Fatalf("claims = %d, want a retry after the error", n)
+	}
+	out := logs.String()
+	if !strings.Contains(out, "claim job failed; retrying") {
+		t.Fatalf("claim error not logged: %s", out)
+	}
+	if strings.Contains(out, sentinelSecret) {
+		t.Fatalf("claim error text logged: %s", out)
+	}
+}
+
+func TestRunnerEnqueueByKindName(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	r := newRunner(t, st, 1)
+	register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
+		return countResult{Done: p.Steps}, nil
+	})
+	ctx := context.Background()
+
+	if _, err := r.Enqueue(ctx, "no_such_kind", json.RawMessage(`{}`)); !errors.Is(err, ErrUnknownKind) {
+		t.Errorf("unknown kind error = %v, want ErrUnknownKind", err)
+	}
+	for _, payload := range []string{`{"steps":"three"}`, `{"other":1}`, `[]`, ``} {
+		if _, err := r.Enqueue(ctx, "count", json.RawMessage(payload)); !errors.Is(err, ErrInvalidPayload) {
+			t.Errorf("payload %q error = %v, want ErrInvalidPayload", payload, err)
+		}
+	}
+	var n int
+	if err := st.Reader.QueryRow("SELECT COUNT(*) FROM jobs").Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 0 {
+		t.Errorf("refused enqueues stored %d jobs, want 0", n)
+	}
+
+	id, err := r.Enqueue(ctx, "count", json.RawMessage(`{"steps":3}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	start(t, r)
+	job := waitFor(t, r, id, finished)
+	if job.Kind != "count" || job.Status != StatusSucceeded || string(job.Result) != `{"done":3}` {
+		t.Errorf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result)
+	}
+}

internal/server/jobs.go

@@ -0,0 +1,107 @@
+package server
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+
+	"github.com/go-chi/chi/v5"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+)
+
+// maxJobBody bounds a POST /jobs request body.
+const maxJobBody = 64 << 10
+
+// jobRequest is the POST /jobs body. Payload is required; a job kind that
+// takes no input is sent the empty object {}.
+type jobRequest struct {
+	Kind    string          `json:"kind"`
+	Payload json.RawMessage `json:"payload"`
+}
+
+// jobCreated is the POST /jobs response.
+type jobCreated struct {
+	ID string `json:"id"`
+}
+
+// postJob serves POST /api/v1/jobs: it enqueues a job of a registered kind
+// and returns 202 with its id, which GET /api/v1/jobs/{id} reports on.
+func postJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		req, err := decodeJobRequest(http.MaxBytesReader(w, r.Body, maxJobBody))
+		if err != nil {
+			writeError(w, r, logger, http.StatusBadRequest, "invalid_request", err.Error())
+			return
+		}
+		id, err := runner.Enqueue(r.Context(), req.Kind, req.Payload)
+		switch {
+		case errors.Is(err, jobs.ErrUnknownKind):
+			writeError(w, r, logger, http.StatusBadRequest, "unknown_kind", "unknown job kind")
+			return
+		case errors.Is(err, jobs.ErrInvalidPayload):
+			writeError(w, r, logger, http.StatusBadRequest, "invalid_payload", "payload does not match the job kind")
+			return
+		case err != nil:
+			logger.ErrorContext(r.Context(), "enqueue job", "kind", req.Kind, "error", err)
+			writeError(w, r, logger, http.StatusInternalServerError, "internal", "internal error")
+			return
+		}
+		writeJSON(w, r, logger, http.StatusAccepted, jobCreated{ID: id})
+	}
+}
+
+// decodeJobRequest reads a single JSON object with a non-empty kind and a
+// payload, refusing unknown fields, an omitted or null payload and trailing
+// data.
+func decodeJobRequest(body io.Reader) (jobRequest, error) {
+	data, err := io.ReadAll(body)
+	if err != nil {
+		var tooLarge *http.MaxBytesError
+		if errors.As(err, &tooLarge) {
+			return jobRequest{}, fmt.Errorf("request body exceeds %d bytes", tooLarge.Limit)
+		}
+		return jobRequest{}, errors.New("cannot read request body")
+	}
+	var req jobRequest
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&req); err != nil {
+		return jobRequest{}, errors.New("request body must be a JSON object with kind and payload")
+	}
+	if _, err := dec.Token(); err != io.EOF {
+		return jobRequest{}, errors.New("request body must be a single JSON object")
+	}
+	if req.Kind == "" {
+		return jobRequest{}, errors.New("kind: required")
+	}
+	if req.Payload == nil {
+		return jobRequest{}, errors.New("payload: required")
+	}
+	if bytes.Equal(req.Payload, []byte("null")) {
+		return jobRequest{}, errors.New("payload: must not be null")
+	}
+	return req, nil
+}
+
+// getJob serves GET /api/v1/jobs/{id}: the job's id, kind, status, latest
+// result and error.
+func getJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		job, err := runner.Get(r.Context(), chi.URLParam(r, "id"))
+		if errors.Is(err, jobs.ErrNotFound) {
+			writeError(w, r, logger, http.StatusNotFound, "not_found", "job not found")
+			return
+		}
+		if err != nil {
+			logger.ErrorContext(r.Context(), "get job", "error", err)
+			writeError(w, r, logger, http.StatusInternalServerError, "internal", "internal error")
+			return
+		}
+		writeJSON(w, r, logger, http.StatusOK, job)
+	}
+}

internal/server/jobs_test.go

@@ -0,0 +1,246 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"maps"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+)
+
+type draftPayload struct{}
+
+type draftResult struct {
+	Days int `json:"days"`
+}
+
+func (d draftResult) PublicJSON() ([]byte, error) { return json.Marshal(d) }
+
+// startedRunner registers a "draft" kind that publishes {"days":1}, waits
+// for release and returns {"days":7}, or fails with a *jobs.Failure when
+// fail is closed first. It starts the runner and stops it when the test ends.
+func startedRunner(t *testing.T, release, fail <-chan struct{}) (*jobs.Runner, *jobs.Kind[draftPayload]) {
+	t.Helper()
+	runner := newTestRunner(t)
+	kind, err := jobs.Register(runner, "draft", func(ctx context.Context, _ draftPayload, publish func(draftResult) error) (draftResult, error) {
+		if err := publish(draftResult{Days: 1}); err != nil {
+			return draftResult{}, err
+		}
+		select {
+		case <-release:
+			return draftResult{Days: 7}, nil
+		case <-fail:
+			return draftResult{}, &jobs.Failure{Code: "no_regulars"}
+		case <-ctx.Done():
+			return draftResult{}, ctx.Err()
+		}
+	}, "no_regulars")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	if err := runner.Start(ctx); err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		cancel()
+		runner.Wait()
+	})
+	return runner, kind
+}
+
+// pollJob GETs the job until want accepts the body, failing after five
+// seconds.
+func pollJob(t *testing.T, h http.Handler, id string, want func(map[string]any) bool) map[string]any {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		rec, body := serve(t, h, signedIn(http.MethodGet, "/api/v1/jobs/"+id, nil))
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body)
+		}
+		if want(body) {
+			if keys := slices.Sorted(maps.Keys(body)); !slices.Equal(keys, []string{"error", "id", "kind", "result", "status"}) {
+				t.Errorf("body fields = %v, want id, kind, status, result, error", keys)
+			}
+			return body
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("job %s never reached the expected state; last %v", id, body)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func resultDays(body map[string]any) float64 {
+	result, _ := body["result"].(map[string]any)
+	days, _ := result["days"].(float64)
+	return days
+}
+
+func TestGetJobShowsProgressThenResult(t *testing.T) {
+	t.Parallel()
+	release := make(chan struct{})
+	runner, kind := startedRunner(t, release, nil)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+	id, err := kind.Enqueue(context.Background(), draftPayload{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	mid := pollJob(t, router, id, func(b map[string]any) bool { return resultDays(b) == 1 })
+	if mid["id"] != id || mid["kind"] != "draft" || mid["status"] != "running" || mid["error"] != nil {
+		t.Errorf("mid-run body = %v, want running draft job %s with no error", mid, id)
+	}
+
+	close(release)
+	done := pollJob(t, router, id, func(b map[string]any) bool { return b["status"] == "succeeded" })
+	if resultDays(done) != 7 || done["error"] != nil {
+		t.Errorf("finished body = %v, want result {\"days\":7} and no error", done)
+	}
+}
+
+func TestGetJobShowsFailure(t *testing.T) {
+	t.Parallel()
+	fail := make(chan struct{})
+	close(fail)
+	runner, kind := startedRunner(t, nil, fail)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+	id, err := kind.Enqueue(context.Background(), draftPayload{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	body := pollJob(t, router, id, func(b map[string]any) bool { return b["status"] == "failed" })
+	if body["error"] != "no_regulars" {
+		t.Errorf("error = %v, want the Failure code", body["error"])
+	}
+}
+
+func TestGetJobUnknownIDIsNotFound(t *testing.T) {
+	t.Parallel()
+	router := newTestRouter(t, "dev")
+
+	rec, _ := serve(t, router, signedIn(http.MethodGet, "/api/v1/jobs/"+store.NewID(), nil))
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("status = %d, want 404", rec.Code)
+	}
+	var body errorBody
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatal(err)
+	}
+	if want := (errorBody{Error: apiError{Code: "not_found", Message: "job not found"}}); body != want {
+		t.Errorf("body = %+v, want %+v", body, want)
+	}
+}
+
+func TestGetJobRequiresSession(t *testing.T) {
+	t.Parallel()
+	router := newTestRouter(t, "dev")
+
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/jobs/"+store.NewID(), nil))
+	if rec.Code != http.StatusUnauthorized {
+		t.Errorf("status = %d, want 401", rec.Code)
+	}
+}
+
+// TestPostJobThenGetProgressAndResult drives the whole flow over the
+// router: POST enqueues a progressive "draft" job, GET reads its partial
+// result while the handler is held mid-run, then its final result.
+func TestPostJobThenGetProgressAndResult(t *testing.T) {
+	t.Parallel()
+	release := make(chan struct{})
+	runner, _ := startedRunner(t, release, nil)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+
+	rec, body := serve(t, router, signedIn(http.MethodPost, "/api/v1/jobs", strings.NewReader(`{"kind":"draft","payload":{}}`)))
+	if rec.Code != http.StatusAccepted {
+		t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body)
+	}
+	id, _ := body["id"].(string)
+	if keys := slices.Sorted(maps.Keys(body)); !slices.Equal(keys, []string{"id"}) || id == "" {
+		t.Fatalf("body = %v, want only a job id", body)
+	}
+
+	mid := pollJob(t, router, id, func(b map[string]any) bool { return resultDays(b) == 1 })
+	if mid["id"] != id || mid["kind"] != "draft" || mid["status"] != "running" || mid["error"] != nil {
+		t.Errorf("mid-run body = %v, want running draft job %s with no error", mid, id)
+	}
+
+	close(release)
+	done := pollJob(t, router, id, func(b map[string]any) bool { return b["status"] == "succeeded" })
+	if done["id"] != id || resultDays(done) != 7 || done["error"] != nil {
+		t.Errorf("finished body = %v, want result {\"days\":7} and no error", done)
+	}
+}
+
+func TestPostJobRefusesBadRequests(t *testing.T) {
+	t.Parallel()
+	runner, _ := startedRunner(t, nil, nil)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+
+	for _, tc := range []struct {
+		name, body, code, message string
+	}{
+		{"unknown kind", `{"kind":"no_such_kind","payload":{}}`, "unknown_kind", "unknown job kind"},
+		{"payload for another type", `{"kind":"draft","payload":{"days":3}}`, "invalid_payload", "payload does not match the job kind"},
+		{"payload not an object", `{"kind":"draft","payload":[1]}`, "invalid_payload", "payload does not match the job kind"},
+		{"missing payload", `{"kind":"draft"}`, "invalid_request", "payload: required"},
+		{"null payload", `{"kind":"draft","payload":null}`, "invalid_request", "payload: must not be null"},
+		{"missing kind", `{"payload":{}}`, "invalid_request", "kind: required"},
+		{"empty body", ``, "invalid_request", "request body must be a JSON object with kind and payload"},
+		{"not an object", `["draft"]`, "invalid_request", "request body must be a JSON object with kind and payload"},
+		{"unknown field", `{"kind":"draft","payload":{},"extra":1}`, "invalid_request", "request body must be a JSON object with kind and payload"},
+		{"trailing data", `{"kind":"draft","payload":{}} {}`, "invalid_request", "request body must be a single JSON object"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			rec := httptest.NewRecorder()
+			router.ServeHTTP(rec, signedIn(http.MethodPost, "/api/v1/jobs", strings.NewReader(tc.body)))
+			if rec.Code != http.StatusBadRequest {
+				t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body)
+			}
+			var body errorBody
+			if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+				t.Fatal(err)
+			}
+			if want := (errorBody{Error: apiError{Code: tc.code, Message: tc.message}}); body != want {
+				t.Errorf("body = %+v, want %+v", body, want)
+			}
+		})
+	}
+}
+
+func TestPostJobNeedsSessionAndCSRF(t *testing.T) {
+	t.Parallel()
+	router := newTestRouter(t, "dev")
+	const body = `{"kind":"noop","payload":{}}`
+
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(body)))
+	if rec.Code != http.StatusUnauthorized || rec.Body.String() != unauthenticatedJSON {
+		t.Errorf("no session: %d %s, want 401 %s", rec.Code, rec.Body, unauthenticatedJSON)
+	}
+
+	for _, header := range []string{"", "wrong-token"} {
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(body))
+		req.AddCookie(validSessionCookie)
+		if header != "" {
+			req.Header.Set(CSRFHeader, header)
+		}
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, req)
+		if rec.Code != http.StatusForbidden || rec.Body.String() != csrfJSON {
+			t.Errorf("CSRF header %q: %d %s, want 403 %s", header, rec.Code, rec.Body, csrfJSON)
+		}
+	}
+}

internal/server/login_test.go

@@ -83,7 +83,7 @@ func newLoginRouter(t *testing.T, issuer string) http.Handler {
 		t.Fatal(err)
 	}
 	r, err := NewRouter(logger, "dev", trusted, flow, sessions, newTestSettings(t),
-		fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
+		newTestRunner(t), fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/server/origin_test.go

@@ -45,7 +45,7 @@ func newOriginRouter(t *testing.T, logs io.Writer) http.Handler {
 	}
 	logger := slog.New(forwarded.LogHandler(slog.NewJSONHandler(logs, nil)))
 	r, err := NewRouter(logger, "dev", trusted, probeAuth{}, stubSessions{}, newTestSettings(t),
-		fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
+		newTestRunner(t), fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/server/server.go

@@ -13,6 +13,7 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/forwarded"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/frontend"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 )
 
 // reservedPrefixes are owned by the API router and are never answered with the
@@ -28,13 +29,17 @@ type RouteMounter interface {
 // has its client address and scheme resolved, honouring forwarded headers
 // only from a peer in trusted (see resolveOrigin). version is reported by
 // GET /health; auth mounts the /auth routes; sessions guards /api/v1 and
-// /media; settings serves /api/v1/settings/household; assets is the web build
+// /media; settings serves /api/v1/settings/household; runner serves
+// POST /api/v1/jobs and GET /api/v1/jobs/{id}; assets is the web build
 // served for every other GET. /health, /auth and the web build are public.
 // Requests to /api/v1 other than GET also need the session's CSRF token in CSRFHeader.
-func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, auth RouteMounter, sessions SessionLookup, settings *household.SettingsService, assets fs.FS) (http.Handler, error) {
+func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, auth RouteMounter, sessions SessionLookup, settings *household.SettingsService, runner *jobs.Runner, assets fs.FS) (http.Handler, error) {
 	if settings == nil {
 		return nil, errors.New("server: nil household settings service")
 	}
+	if runner == nil {
+		return nil, errors.New("server: nil job runner")
+	}
 	notFound := apiErrorHandler(logger, http.StatusNotFound, "not_found", "not found")
 	methodNotAllowed := apiErrorHandler(logger, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
 
@@ -63,6 +68,8 @@ func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, a
 	apiV1.Get("/me", me(logger))
 	apiV1.Get("/settings/household", getHouseholdSettings(logger, settings))
 	apiV1.Patch("/settings/household", patchHouseholdSettings(logger, settings))
+	apiV1.Post("/jobs", postJob(logger, runner))
+	apiV1.Get("/jobs/{id}", getJob(logger, runner))
 	r.Route("/api", func(r chi.Router) {
 		r.MethodNotAllowed(methodNotAllowed)
 		r.NotFound(notFound)

internal/server/server_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/go-chi/chi/v5"
 
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
 )
@@ -27,17 +28,17 @@ const (
 
 func newTestRouter(t *testing.T, version string) http.Handler {
 	t.Helper()
-	return newTestRouterWith(t, version, newTestSettings(t))
+	return newTestRouterWith(t, version, newTestSettings(t), newTestRunner(t))
 }
 
-func newTestRouterWith(t *testing.T, version string, settings *household.SettingsService) http.Handler {
+func newTestRouterWith(t *testing.T, version string, settings *household.SettingsService, runner *jobs.Runner) http.Handler {
 	t.Helper()
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 	assets := fstest.MapFS{
 		"index.html":             {Data: []byte(fixtureIndex)},
 		"assets/index-AbC123.js": {Data: []byte(fixtureAsset)},
 	}
-	r, err := NewRouter(logger, version, nil, stubAuth{}, stubSessions{}, settings, assets)
+	r, err := NewRouter(logger, version, nil, stubAuth{}, stubSessions{}, settings, runner, assets)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -70,9 +71,8 @@ func TestAuthRoutesMounted(t *testing.T) {
 	}
 }
 
-// newTestSettings returns a settings service over a migrated temp database
-// seeded with zone America/New_York.
-func newTestSettings(t *testing.T) *household.SettingsService {
+// newTestStore returns a migrated temp database, closed when the test ends.
+func newTestStore(t *testing.T) *store.Store {
 	t.Helper()
 	ctx := context.Background()
 	st, err := store.Open(ctx, t.TempDir())
@@ -87,21 +87,49 @@ func newTestSettings(t *testing.T) *household.SettingsService {
 	if err := store.Migrate(ctx, st.Writer, migrations.FS); err != nil {
 		t.Fatal(err)
 	}
-	if err := household.SeedSettings(ctx, st, "America/New_York"); err != nil {
+	return st
+}
+
+// newTestSettings returns a settings service over a migrated temp database
+// seeded with zone America/New_York.
+func newTestSettings(t *testing.T) *household.SettingsService {
+	t.Helper()
+	st := newTestStore(t)
+	if err := household.SeedSettings(context.Background(), st, "America/New_York"); err != nil {
 		t.Fatal(err)
 	}
 	return household.NewSettingsService(st)
 }
 
+// newTestRunner returns an unstarted job runner over a migrated temp
+// database.
+func newTestRunner(t *testing.T) *jobs.Runner {
+	t.Helper()
+	runner, err := jobs.NewRunner(newTestStore(t), slog.New(slog.NewTextHandler(io.Discard, nil)), 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return runner
+}
+
 func TestNewRouterRequiresSettings(t *testing.T) {
 	t.Parallel()
 
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
-	if _, err := NewRouter(logger, "dev", nil, stubAuth{}, stubSessions{}, nil, fstest.MapFS{"index.html": {}}); err == nil {
+	if _, err := NewRouter(logger, "dev", nil, stubAuth{}, stubSessions{}, nil, newTestRunner(t), fstest.MapFS{"index.html": {}}); err == nil {
 		t.Fatal("NewRouter accepted a nil settings service")
 	}
 }
 
+func TestNewRouterRequiresJobRunner(t *testing.T) {
+	t.Parallel()
+
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	if _, err := NewRouter(logger, "dev", nil, stubAuth{}, stubSessions{}, newTestSettings(t), nil, fstest.MapFS{"index.html": {}}); err == nil {
+		t.Fatal("NewRouter accepted a nil job runner")
+	}
+}
+
 func TestHealth(t *testing.T) {
 	t.Parallel()
 

internal/server/settings_test.go

@@ -84,7 +84,7 @@ func TestGetHouseholdSettings(t *testing.T) {
 func TestPatchHouseholdSettings(t *testing.T) {
 	t.Parallel()
 	settings := newTestSettings(t)
-	router := newTestRouterWith(t, "dev", settings)
+	router := newTestRouterWith(t, "dev", settings, newTestRunner(t))
 
 	var notified []household.Settings
 	settings.Subscribe(func(s household.Settings) { notified = append(notified, s) })
@@ -146,7 +146,7 @@ func TestPatchHouseholdSettingsRejected(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
 			settings := newTestSettings(t)
-			router := newTestRouterWith(t, "dev", settings)
+			router := newTestRouterWith(t, "dev", settings, newTestRunner(t))
 			notified := 0
 			settings.Subscribe(func(household.Settings) { notified++ })
 
@@ -216,7 +216,7 @@ func TestHouseholdSettingsWithRealSession(t *testing.T) {
 	}
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 	router, err := NewRouter(logger, "dev", nil, stubAuth{}, sessions, household.NewSettingsService(st),
-		fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
+		newTestRunner(t), fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/store/migrate_test.go

@@ -175,8 +175,8 @@ func TestMigrateEmbeddedMigrations(t *testing.T) {
 	if err := Migrate(context.Background(), s.Writer, migrations.FS); err != nil {
 		t.Fatal(err)
 	}
-	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by" {
-		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by", got)
+	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status" {
+		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status", got)
 	}
 }
 
@@ -196,8 +196,8 @@ func TestMigrateBootstrapThenEmbeddedInitApplies(t *testing.T) {
 	if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
 		t.Fatal(err)
 	}
-	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by" {
-		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by", got)
+	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status" {
+		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status", got)
 	}
 	if !tableExists(t, s, "dinners") {
 		t.Error("table from 0001_init was not created")
@@ -262,3 +262,50 @@ func TestMigrateReopenSameDataDirIsNoOp(t *testing.T) {
 		t.Errorf("schema changed on reopen:\n got %v\nwant %v", afterSchema, beforeSchema)
 	}
 }
+
+// TestMigrateJobsStatusMapsLegacyRows upgrades a database holding jobs in
+// every 0001 status and checks 0003 carries each row over with its status
+// mapped and every other column unchanged.
+func TestMigrateJobsStatusMapsLegacyRows(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+	before := fstest.MapFS{}
+	for _, name := range []string{"0000_bootstrap.sql", "0001_init.sql", "0002_household_settings_updated_by.sql"} {
+		body, err := fs.ReadFile(migrations.FS, name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		before[name] = &fstest.MapFile{Data: body}
+	}
+	if err := Migrate(ctx, s.Writer, before); err != nil {
+		t.Fatal(err)
+	}
+	legacy := map[string]string{"pending": "queued", "running": "running", "done": "succeeded", "failed": "failed"}
+	for old := range legacy {
+		if _, err := s.Writer.Exec(`INSERT INTO jobs (id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at)
+			VALUES (?, 'suggest', ?, '{"n":1}', '{"r":2}', 'boom', 's', 'f', 'c', 'u')`, "job-"+old, old); err != nil {
+			t.Fatalf("insert %s job: %v", old, err)
+		}
+	}
+
+	if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	for old, want := range legacy {
+		var kind, status, payload, result, errText, startedAt, finishedAt, createdAt, updatedAt string
+		if err := s.Reader.QueryRow(`SELECT kind, status, payload, result, error, started_at, finished_at, created_at, updated_at FROM jobs WHERE id = ?`, "job-"+old).
+			Scan(&kind, &status, &payload, &result, &errText, &startedAt, &finishedAt, &createdAt, &updatedAt); err != nil {
+			t.Fatalf("read %s job: %v", old, err)
+		}
+		if status != want {
+			t.Errorf("%s job status = %q, want %q", old, status, want)
+		}
+		got := []string{kind, payload, result, errText, startedAt, finishedAt, createdAt, updatedAt}
+		if wantCols := []string{"suggest", `{"n":1}`, `{"r":2}`, "boom", "s", "f", "c", "u"}; !slices.Equal(got, wantCols) {
+			t.Errorf("%s job columns = %v, want %v", old, got, wantCols)
+		}
+	}
+	if tableExists(t, s, "jobs_new") {
+		t.Error("jobs_new left behind")
+	}
+}

internal/store/schema_test.go

@@ -165,7 +165,7 @@ func testRow(table string, i int) map[string]any {
 		r["attempts"] = 0
 	case "jobs":
 		r["kind"] = "suggest"
-		r["status"] = "pending"
+		r["status"] = "queued"
 		r["payload"] = "{}"
 	case "schedule_runs":
 		r["schedule"] = "plan_draft"
@@ -330,7 +330,7 @@ func TestSchemaEnumChecksRejectInvalidValues(t *testing.T) {
 		{"notification_prefs", "chat", 2},
 		{"notification_outbox", "channel", "email"},
 		{"notification_outbox", "status", "queued"},
-		{"jobs", "status", "queued"},
+		{"jobs", "status", "bogus"},
 		{"schedule_runs", "schedule", "hourly"},
 	}
 	s := seededStore(t)

migrations/0003_jobs_status.sql

@@ -0,0 +1,32 @@
+-- 0003_jobs_status: jobs.status becomes queued, running, succeeded, failed,
+-- the states the job runner (internal/jobs) reports. SQLite cannot change a
+-- CHECK in place, so the table is rebuilt: create the new table, copy every
+-- row, drop the old one and rename. Nothing references jobs and it has no
+-- indexes or triggers, so the rebuild touches no other schema object. Rows
+-- written under the 0001 values are mapped (pending -> queued,
+-- done -> succeeded) so the copy cannot fail on an existing database.
+
+CREATE TABLE jobs_new (
+	id          TEXT NOT NULL PRIMARY KEY,
+	kind        TEXT NOT NULL,
+	status      TEXT NOT NULL CHECK (status IN ('queued', 'running', 'succeeded', 'failed')),
+	payload     TEXT NOT NULL, -- JSON
+	result      TEXT,          -- JSON, written progressively
+	error       TEXT,
+	started_at  TEXT,
+	finished_at TEXT,
+	created_at  TEXT NOT NULL,
+	updated_at  TEXT NOT NULL
+);
+
+INSERT INTO jobs_new (id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at)
+SELECT
+	id,
+	kind,
+	CASE status WHEN 'pending' THEN 'queued' WHEN 'done' THEN 'succeeded' ELSE status END,
+	payload, result, error, started_at, finished_at, created_at, updated_at
+FROM jobs;
+
+DROP TABLE jobs;
+
+ALTER TABLE jobs_new RENAME TO jobs;

migrations/queries/jobs.sql

@@ -0,0 +1,37 @@
+-- name: CreateJob :exec
+INSERT INTO jobs (id, kind, status, payload, created_at, updated_at)
+VALUES (?, ?, 'queued', ?, ?, ?);
+
+-- name: ClaimNextJob :one
+-- Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
+-- when no job is queued.
+UPDATE jobs
+SET status = 'running', started_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE id = (SELECT id FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 1)
+RETURNING *;
+
+-- name: PublishJobResult :execrows
+-- Stores a partial result of a running job.
+UPDATE jobs
+SET result = ?, updated_at = ?
+WHERE id = ? AND status = 'running';
+
+-- name: SucceedJob :execrows
+UPDATE jobs
+SET status = 'succeeded', result = sqlc.arg(result), finished_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'running';
+
+-- name: FailJob :execrows
+UPDATE jobs
+SET status = 'failed', error = sqlc.arg(error), finished_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'running';
+
+-- name: FailRunningJobs :execrows
+-- Fails every job left running, at boot before any worker starts.
+UPDATE jobs
+SET status = 'failed', error = sqlc.arg(error), finished_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE status = 'running';
+
+-- name: GetJob :one
+SELECT * FROM jobs
+WHERE id = ?;