diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 750478f..0d77b3d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -159,9 +159,18 @@ automatic** (D-021). is always the key the provider sends. Request and response bodies are redacted before they are stored: the decrypted API key, Authorization and API-key header values, Google Chat webhook URLs and URLs - with key-like query parameters are removed, inside JSON strings too. + with key-like query parameters are removed, inside JSON strings too; the + skill, provider and model columns are redacted the same way. Writing the + row is bounded: a run whose work ended in time must write its row before + the skill's deadline or it fails with `llm.ErrTimeout` and its row records + the timeout; a failed run's row has a further fixed bound, and if even + that write fails the run returns the write error with its own and no row + is stored. A stored row's status therefore always matches what the caller + received. `harness.Purge` deletes rows older than `harness.RunRetention` (30 days), - clearing the `suggestion_batches` and `week_plans` references first. + clearing the `suggestion_batches` and `week_plans` references first. The + harness does not call it: the scheduler's daily `ai_runs` purge + (Feature 20) must call `harness.Purge` with `now - RunRetention`. - **Concurrency guards** (`harness.Guard`, one per process): `harness.Run` takes a scope key, and concurrent runs of one skill with one scope share a single in-flight run and its result, error included (single-flight); runs @@ -221,7 +230,7 @@ purge are fixed and not configurable. | `day_of_reminder` | Daily 16:30 | Tonight's dinner reminder. | | `rotation_check` | Daily 09:00 | Asks "Keep it in rotation?" for tried dinners. | | `week_rollover` | Monday 00:05 (fixed) | Freezes last week's Supply run into the Supply archive. | -| `ai_runs` purge | Daily (fixed) | Deletes `ai_runs` rows past 30-day retention. | +| `ai_runs` purge | Daily (fixed) | Calls `harness.Purge` to delete `ai_runs` rows past 30-day retention (Feature 20). | Runs are recorded in `schedule_runs` and executed through `jobs`. diff --git a/internal/ai/harness/guard_test.go b/internal/ai/harness/guard_test.go index e292526..04eecd7 100644 --- a/internal/ai/harness/guard_test.go +++ b/internal/ai/harness/guard_test.go @@ -300,8 +300,9 @@ func TestRunTimeout(t *testing.T) { cfg := testConfig(t, &f) f.QueueGenerate(reply(`{"echo":"taco"}`)) timeout := 100 * time.Millisecond - // Holding the store's only writer connection makes the save wait - // past the deadline, after the work itself has ended in time. + // Holding the store's only writer connection past the deadline + // keeps the ok row from being written in time, after the work + // itself has ended in time. conn, err := cfg.Store.Writer.Conn(context.Background()) if err != nil { t.Fatal(err) @@ -316,11 +317,39 @@ func TestRunTimeout(t *testing.T) { t.Fatalf("out = %+v, want no output", out) } assertCalls(t, &f, 1) - // An ok row with no error proves the work ended in time and only - // the save ran past the deadline. + // The row records the timeout the caller received, not the ok + // outcome the work had. rows := runRows(t, cfg.Store) - if len(rows) != 1 || rows[0].Status != string(StatusOK) || rows[0].Error.Valid { - t.Fatalf("rows = %+v, want one ok row with no error", rows) + if len(rows) != 1 || rows[0].Status != string(StatusProviderError) || !strings.Contains(rows[0].Error.String, llm.ErrTimeout.Error()) { + t.Fatalf("rows = %+v, want one provider_error row naming the timeout", rows) + } + }) + t.Run("writer blocked", func(t *testing.T) { + var f llm.Fake + cfg := testConfig(t, &f) + f.QueueGenerate(reply(`{"echo":"taco"}`)) + timeout := 50 * time.Millisecond + saveTimeout := runSaveTimeout + runSaveTimeout = 100 * time.Millisecond + defer func() { runSaveTimeout = saveTimeout }() + // The writer stays held until Run has returned, so neither save + // can be written. + conn, err := cfg.Store.Writer.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + started := time.Now() + _, err = Run(context.Background(), cfg, timeoutSkill{timeout: timeout}, "week-1", echoInput{Word: "taco"}) + elapsed := time.Since(started) + conn.Close() + if !errors.Is(err, llm.ErrTimeout) || !strings.Contains(err.Error(), "record ai_runs row") { + t.Fatalf("err = %v, want ErrTimeout and the row write failure", err) + } + if bound := timeout + runSaveTimeout + time.Second; elapsed > bound { + t.Fatalf("Run took %v with the writer blocked, want at most %v", elapsed, bound) + } + if rows := runRows(t, cfg.Store); len(rows) != 0 { + t.Fatalf("rows = %+v, want none", rows) } }) t.Run("waiting for a slot", func(t *testing.T) { diff --git a/internal/ai/harness/harness.go b/internal/ai/harness/harness.go index 0ec9ee3..c1d7296 100644 --- a/internal/ai/harness/harness.go +++ b/internal/ai/harness/harness.go @@ -42,8 +42,8 @@ type Skill[In, Out any] interface { // is validated again; problems found then are repaired like any other. // Its error fails the run without a repair turn. PostProcess(out Out) (Out, error) - // Timeout bounds a whole run, waiting for LLM call slots and the - // repair turn included. It is DefaultTimeout unless the skill needs + // Timeout bounds a whole run, waiting for LLM call slots, the repair + // turn and writing a successful run's ai_runs row included (see Run). It is DefaultTimeout unless the skill needs // another, and must be positive. Timeout() time.Duration } @@ -136,6 +136,15 @@ func (e *BadOutputError) Unwrap() error { return llm.ErrBadOutput } // before anything else, and writes no row since the row cannot be // written without one. When the row cannot be written the run fails with // that error too, even if its output was usable. +// +// Writing the row is bounded too. A run whose work ended in time must +// write its row before s.Timeout() as well; if the store's writer is not +// free by then, the run fails with llm.ErrTimeout and its row records the +// timeout instead. The row of a run that failed is written within a +// further fixed bound (runSaveTimeout); if it cannot be, the run fails with +// the write error alongside its own, and no row is stored. So Run returns +// at most that bound after s.Timeout(), and a stored row's status always +// matches what the caller received. func Run[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], scope string, in In) (Out, error) { var zero Out if err := cfg.validate(); err != nil { @@ -168,7 +177,7 @@ func runOnce[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], tim deadline := time.Now().Add(timeout) ctx, cancel := context.WithDeadlineCause(ctx, deadline, expired) defer cancel() - log := &runLog{start: store.Now()} + log := &runLog{id: store.NewID(), start: store.Now()} out, status, err := execute(ctx, cfg, s, in, log) // Work that only ends after the deadline, such as a slow PostProcess, // may return no error; it is still a timeout, not a success. The clock @@ -180,19 +189,34 @@ func runOnce[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], tim err = fmt.Errorf("%w: %w", expired, err) } } - if saveErr := log.save(ctx, cfg, s.Name(), s.Version(), status, err); saveErr != nil { + // The saves ignore the callers' cancellation, so a cancelled run is + // recorded too, but each is bounded. Work that ended in time must also + // be recorded in time: its row is written within the deadline or not at + // all, so a stored ok or repaired row always means the caller got the + // output. + if err == nil { + sctx, cancel := context.WithDeadline(context.WithoutCancel(ctx), deadline) + saveErr := log.save(sctx, cfg, s.Name(), s.Version(), status, nil) + late := sctx.Err() != nil + cancel() + if saveErr == nil { + return out, nil + } + if !late { + return zero, fmt.Errorf("harness: skill %s v%d: record ai_runs row: %w", s.Name(), s.Version(), saveErr) + } + // The store's writer was not free before the deadline, so the run + // has not ended in time. + err, status = expired, StatusProviderError + } + // A run that failed is recorded within runSaveTimeout; when it cannot + // be, the run fails with both errors. + sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), runSaveTimeout) + defer cancel() + if saveErr := log.save(sctx, cfg, s.Name(), s.Version(), status, err); saveErr != nil { return zero, errors.Join(err, fmt.Errorf("harness: skill %s v%d: record ai_runs row: %w", s.Name(), s.Version(), saveErr)) } - if err != nil { - return zero, err - } - // The save may wait past the deadline for the store's single writer. - // The run has then not ended in time, so it fails, though its row - // records the work's outcome as it stood before the save. - if !time.Now().Before(deadline) { - return zero, expired - } - return out, nil + return zero, err } // execute renders the prompt, builds the provider and sends the turns, diff --git a/internal/ai/harness/runs.go b/internal/ai/harness/runs.go index fb12141..5bf2e86 100644 --- a/internal/ai/harness/runs.go +++ b/internal/ai/harness/runs.go @@ -38,9 +38,17 @@ const ( // Purge with time.Now().Add(-RunRetention). const RunRetention = 30 * 24 * time.Hour +// runSaveTimeout bounds writing the ai_runs row of a run that has failed, +// timed out or been cancelled, so a blocked writer cannot hold Run up +// indefinitely. A variable so tests can shorten it. +var runSaveTimeout = 5 * time.Second + // Purge deletes the ai_runs rows created before olderThan and returns how // many it deleted. The suggestion_batches and week_plans references to // those rows are cleared first, in the same transaction. +// +// Nothing in this package calls Purge: the scheduler's daily ai_runs purge +// (Feature 20) must call it with time.Now().Add(-RunRetention). func Purge(ctx context.Context, st *store.Store, olderThan time.Time) (int64, error) { if olderThan.IsZero() { return 0, errors.New("harness: purge ai_runs: zero cutoff") @@ -67,6 +75,9 @@ func Purge(ctx context.Context, st *store.Store, olderThan time.Time) (int64, er // runLog collects one run's turns for its ai_runs row. type runLog struct { + // id is the row's ID, fixed for the run so that a second save attempt + // can never add a second row. + id string start store.Timestamp requests []loggedRequest responses []loggedResponse @@ -138,8 +149,8 @@ func (l *runLog) add(apiKey string, req llm.Request, resp llm.Response, err erro l.tokensOut += int64(resp.Usage.OutputTokens) } -// save writes the run's ai_runs row. It is written even when ctx is done, -// so a cancelled run is recorded too. +// save writes the run's ai_runs row within ctx, which the caller bounds. +// Every stored string is redacted. func (l *runLog) save(ctx context.Context, cfg Config, skill string, version int, status RunStatus, runErr error) error { // A run that failed before its first turn stores an empty list. request, err := marshalRedacted(nonNil(l.requests), cfg.APIKey) @@ -158,12 +169,12 @@ func (l *runLog) save(ctx context.Context, cfg Config, skill string, version int errText = sql.NullString{String: redact("first reply: "+strings.Join(l.problems, "; "), cfg.APIKey), Valid: true} } now := store.Now() - return db.New(cfg.Store.Writer).CreateAIRun(context.WithoutCancel(ctx), db.CreateAIRunParams{ - ID: store.NewID(), - Skill: skill, + return db.New(cfg.Store.Writer).CreateAIRun(ctx, db.CreateAIRunParams{ + ID: l.id, + Skill: redact(skill, cfg.APIKey), SkillVersion: strconv.Itoa(version), - Provider: cfg.Settings.Provider, - Model: cfg.Settings.Model, + Provider: redact(cfg.Settings.Provider, cfg.APIKey), + Model: redact(cfg.Settings.Model, cfg.APIKey), Temperature: cfg.Settings.Temperature, Status: string(status), LatencyMs: sql.NullInt64{Int64: now.Sub(l.start.Time).Milliseconds(), Valid: true}, diff --git a/internal/ai/harness/runs_test.go b/internal/ai/harness/runs_test.go index 963207a..64d7107 100644 --- a/internal/ai/harness/runs_test.go +++ b/internal/ai/harness/runs_test.go @@ -276,6 +276,9 @@ func TestRunRowsHoldNoSecrets(t *testing.T) { return &f, nil } cfg.Persona = "Our kitchen. " + leaky + // A model value can hold a credential too: every stored + // column is redacted, not only the bodies. + cfg.Settings.Model = "model-" + seededKey + " " + seededWebhook _, _ = Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "tacos " + seededWebhook}) assertCalls(t, &f, len(script)) @@ -289,6 +292,9 @@ func TestRunRowsHoldNoSecrets(t *testing.T) { t.Errorf("stored row holds %q:\n%s", s, text) } } + if want := "model-" + redactionMarker + " " + redactionMarker; rows[0].Model != want { + t.Errorf("stored model = %q, want %q", rows[0].Model, want) + } for _, kept := range []string{"Our kitchen.", "Repeat: tacos", redactionMarker} { if !strings.Contains(text, kept) { t.Errorf("stored row lacks %q:\n%s", kept, text)