Bound ai_runs saves and redact every stored column
- runOnce no longer saves with an unbounded context.WithoutCancel. Work that ended in time must write its row before the skill deadline; if the writer is not free by then the run fails with llm.ErrTimeout and its row records the timeout. A failed run's row is written within runSaveTimeout (5s); if that write fails too, the run returns both errors and no row is stored. A stored row's status now always matches what the caller received. The row ID is fixed per run, so a second save attempt can never add a second row. - skill, provider and model are redacted with the same redactor as the bodies; TestRunRowsHoldNoSecrets seeds the API key and a webhook URL into the model value. - guard_test: "row saved late" now expects a provider_error timeout row; new "writer blocked" case proves Run returns ErrTimeout within the bound and stores no row. - Purge doc and ARCHITECTURE: the daily purge is owned by Feature 20's scheduler, which must call harness.Purge. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+12
-3
@@ -159,9 +159,18 @@ automatic** (D-021).
|
|||||||
is always the key the provider sends. Request and response
|
is always the key the provider sends. Request and response
|
||||||
bodies are redacted before they are stored: the decrypted API key,
|
bodies are redacted before they are stored: the decrypted API key,
|
||||||
Authorization and API-key header values, Google Chat webhook URLs and URLs
|
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),
|
`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`
|
- **Concurrency guards** (`harness.Guard`, one per process): `harness.Run`
|
||||||
takes a scope key, and concurrent runs of one skill with one scope share a
|
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
|
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. |
|
| `day_of_reminder` | Daily 16:30 | Tonight's dinner reminder. |
|
||||||
| `rotation_check` | Daily 09:00 | Asks "Keep it in rotation?" for tried dinners. |
|
| `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. |
|
| `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`.
|
Runs are recorded in `schedule_runs` and executed through `jobs`.
|
||||||
|
|
||||||
|
|||||||
@@ -300,8 +300,9 @@ func TestRunTimeout(t *testing.T) {
|
|||||||
cfg := testConfig(t, &f)
|
cfg := testConfig(t, &f)
|
||||||
f.QueueGenerate(reply(`{"echo":"taco"}`))
|
f.QueueGenerate(reply(`{"echo":"taco"}`))
|
||||||
timeout := 100 * time.Millisecond
|
timeout := 100 * time.Millisecond
|
||||||
// Holding the store's only writer connection makes the save wait
|
// Holding the store's only writer connection past the deadline
|
||||||
// past the deadline, after the work itself has ended in time.
|
// 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())
|
conn, err := cfg.Store.Writer.Conn(context.Background())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@@ -316,11 +317,39 @@ func TestRunTimeout(t *testing.T) {
|
|||||||
t.Fatalf("out = %+v, want no output", out)
|
t.Fatalf("out = %+v, want no output", out)
|
||||||
}
|
}
|
||||||
assertCalls(t, &f, 1)
|
assertCalls(t, &f, 1)
|
||||||
// An ok row with no error proves the work ended in time and only
|
// The row records the timeout the caller received, not the ok
|
||||||
// the save ran past the deadline.
|
// outcome the work had.
|
||||||
rows := runRows(t, cfg.Store)
|
rows := runRows(t, cfg.Store)
|
||||||
if len(rows) != 1 || rows[0].Status != string(StatusOK) || rows[0].Error.Valid {
|
if len(rows) != 1 || rows[0].Status != string(StatusProviderError) || !strings.Contains(rows[0].Error.String, llm.ErrTimeout.Error()) {
|
||||||
t.Fatalf("rows = %+v, want one ok row with no error", rows)
|
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) {
|
t.Run("waiting for a slot", func(t *testing.T) {
|
||||||
|
|||||||
@@ -42,8 +42,8 @@ type Skill[In, Out any] interface {
|
|||||||
// is validated again; problems found then are repaired like any other.
|
// is validated again; problems found then are repaired like any other.
|
||||||
// Its error fails the run without a repair turn.
|
// Its error fails the run without a repair turn.
|
||||||
PostProcess(out Out) (Out, error)
|
PostProcess(out Out) (Out, error)
|
||||||
// Timeout bounds a whole run, waiting for LLM call slots and the
|
// Timeout bounds a whole run, waiting for LLM call slots, the repair
|
||||||
// repair turn included. It is DefaultTimeout unless the skill needs
|
// 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.
|
// another, and must be positive.
|
||||||
Timeout() time.Duration
|
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
|
// 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
|
// written without one. When the row cannot be written the run fails with
|
||||||
// that error too, even if its output was usable.
|
// that error too, even if its output was usable.
|
||||||
|
//
|
||||||
|
// Writing the row is bounded too. A run whose work ended in time must
|
||||||
|
// write its row before s.Timeout() as well; if the store's writer is not
|
||||||
|
// free by then, the run fails with llm.ErrTimeout and its row records the
|
||||||
|
// timeout instead. The row of a run that failed is written within a
|
||||||
|
// further fixed bound (runSaveTimeout); if it cannot be, the run fails with
|
||||||
|
// the write error alongside its own, and no row is stored. So Run returns
|
||||||
|
// at most that bound after s.Timeout(), and a stored row's status always
|
||||||
|
// matches what the caller received.
|
||||||
func Run[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], scope string, in In) (Out, error) {
|
func Run[In, Out any](ctx context.Context, cfg Config, s Skill[In, Out], scope string, in In) (Out, error) {
|
||||||
var zero Out
|
var zero Out
|
||||||
if err := cfg.validate(); err != nil {
|
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)
|
deadline := time.Now().Add(timeout)
|
||||||
ctx, cancel := context.WithDeadlineCause(ctx, deadline, expired)
|
ctx, cancel := context.WithDeadlineCause(ctx, deadline, expired)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
log := &runLog{start: store.Now()}
|
log := &runLog{id: store.NewID(), start: store.Now()}
|
||||||
out, status, err := execute(ctx, cfg, s, in, log)
|
out, status, err := execute(ctx, cfg, s, in, log)
|
||||||
// Work that only ends after the deadline, such as a slow PostProcess,
|
// 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
|
// 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)
|
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))
|
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
|
||||||
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
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// execute renders the prompt, builds the provider and sends the turns,
|
// execute renders the prompt, builds the provider and sends the turns,
|
||||||
|
|||||||
@@ -38,9 +38,17 @@ const (
|
|||||||
// Purge with time.Now().Add(-RunRetention).
|
// Purge with time.Now().Add(-RunRetention).
|
||||||
const RunRetention = 30 * 24 * time.Hour
|
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
|
// Purge deletes the ai_runs rows created before olderThan and returns how
|
||||||
// many it deleted. The suggestion_batches and week_plans references to
|
// many it deleted. The suggestion_batches and week_plans references to
|
||||||
// those rows are cleared first, in the same transaction.
|
// 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) {
|
func Purge(ctx context.Context, st *store.Store, olderThan time.Time) (int64, error) {
|
||||||
if olderThan.IsZero() {
|
if olderThan.IsZero() {
|
||||||
return 0, errors.New("harness: purge ai_runs: zero cutoff")
|
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.
|
// runLog collects one run's turns for its ai_runs row.
|
||||||
type runLog struct {
|
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
|
start store.Timestamp
|
||||||
requests []loggedRequest
|
requests []loggedRequest
|
||||||
responses []loggedResponse
|
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)
|
l.tokensOut += int64(resp.Usage.OutputTokens)
|
||||||
}
|
}
|
||||||
|
|
||||||
// save writes the run's ai_runs row. It is written even when ctx is done,
|
// save writes the run's ai_runs row within ctx, which the caller bounds.
|
||||||
// so a cancelled run is recorded too.
|
// Every stored string is redacted.
|
||||||
func (l *runLog) save(ctx context.Context, cfg Config, skill string, version int, status RunStatus, runErr error) error {
|
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.
|
// A run that failed before its first turn stores an empty list.
|
||||||
request, err := marshalRedacted(nonNil(l.requests), cfg.APIKey)
|
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}
|
errText = sql.NullString{String: redact("first reply: "+strings.Join(l.problems, "; "), cfg.APIKey), Valid: true}
|
||||||
}
|
}
|
||||||
now := store.Now()
|
now := store.Now()
|
||||||
return db.New(cfg.Store.Writer).CreateAIRun(context.WithoutCancel(ctx), db.CreateAIRunParams{
|
return db.New(cfg.Store.Writer).CreateAIRun(ctx, db.CreateAIRunParams{
|
||||||
ID: store.NewID(),
|
ID: l.id,
|
||||||
Skill: skill,
|
Skill: redact(skill, cfg.APIKey),
|
||||||
SkillVersion: strconv.Itoa(version),
|
SkillVersion: strconv.Itoa(version),
|
||||||
Provider: cfg.Settings.Provider,
|
Provider: redact(cfg.Settings.Provider, cfg.APIKey),
|
||||||
Model: cfg.Settings.Model,
|
Model: redact(cfg.Settings.Model, cfg.APIKey),
|
||||||
Temperature: cfg.Settings.Temperature,
|
Temperature: cfg.Settings.Temperature,
|
||||||
Status: string(status),
|
Status: string(status),
|
||||||
LatencyMs: sql.NullInt64{Int64: now.Sub(l.start.Time).Milliseconds(), Valid: true},
|
LatencyMs: sql.NullInt64{Int64: now.Sub(l.start.Time).Milliseconds(), Valid: true},
|
||||||
|
|||||||
@@ -276,6 +276,9 @@ func TestRunRowsHoldNoSecrets(t *testing.T) {
|
|||||||
return &f, nil
|
return &f, nil
|
||||||
}
|
}
|
||||||
cfg.Persona = "Our kitchen. " + leaky
|
cfg.Persona = "Our kitchen. " + leaky
|
||||||
|
// A model value can hold a credential too: every stored
|
||||||
|
// column is redacted, not only the bodies.
|
||||||
|
cfg.Settings.Model = "model-" + seededKey + " " + seededWebhook
|
||||||
_, _ = Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "tacos " + seededWebhook})
|
_, _ = Run(context.Background(), cfg, echoSkill{}, "week-1", echoInput{Word: "tacos " + seededWebhook})
|
||||||
assertCalls(t, &f, len(script))
|
assertCalls(t, &f, len(script))
|
||||||
|
|
||||||
@@ -289,6 +292,9 @@ func TestRunRowsHoldNoSecrets(t *testing.T) {
|
|||||||
t.Errorf("stored row holds %q:\n%s", s, text)
|
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} {
|
for _, kept := range []string{"Our kitchen.", "Repeat: tacos", redactionMarker} {
|
||||||
if !strings.Contains(text, kept) {
|
if !strings.Contains(text, kept) {
|
||||||
t.Errorf("stored row lacks %q:\n%s", kept, text)
|
t.Errorf("stored row lacks %q:\n%s", kept, text)
|
||||||
|
|||||||
Reference in New Issue
Block a user