diff --git a/dash/overview.go b/dash/overview.go index a02b717..2090804 100644 --- a/dash/overview.go +++ b/dash/overview.go @@ -55,6 +55,17 @@ type Overview struct { Requests int64 `json:"requests"` Sessions int64 `json:"sessions"` + // InvalidConfigRequests is how many of Requests ran with NO compaction because this + // account's stored configuration failed to build — proxy/tenancy.go's build() fails open + // on purpose (a bad config row must never take someone's agent offline) by forwarding + // uncompacted and marking the row's preset "invalid", but until now that marker was only + // ever visible in the proxy's own log, never on the page an account owner or a manager + // actually looks at. A real incident (#118 removed a config key with no migration for the + // accounts already using it) ran for hours as nine accounts' compaction silently went to + // zero before anyone noticed from a log line rather than from here. Nonzero here means the + // account's Settings page needs attention now, not "check the logs". + InvalidConfigRequests int64 `json:"invalid_config_requests"` + TokensBefore int64 `json:"tokens_before"` TokensAfter int64 `json:"tokens_after"` // SavedGross re-counts the same compaction every turn the agent re-sends the @@ -616,8 +627,18 @@ func (d *DB) Overview(f Filter) (*Overview, error) { keepAlivePingUSD float64 accountingM, cacheMissM, uncompressedM map[string]int64 p95cg, p95up float64 + invalidConfigRequests int64 ) var g errgroup.Group + // See InvalidConfigRequests' own comment: preset is set to "invalid" by + // proxy/tenancy.go's build() exactly when this account's stored configuration failed to + // build, on every request forwarded uncompacted while that lasts. Cheap: cond already + // scopes this to the tenant/window idx_requests_tenant covers, so this is a filter over an + // already-narrow row set, not a fresh scan. + g.Go(func() error { + return d.sql.QueryRow(`SELECT COUNT(*) FROM requests r + WHERE `+cond+` AND r.preset = 'invalid'`, args...).Scan(&invalidConfigRequests) + }) // The replay ceiling's raw form, corrected below by the inflation query once both are in — // see the comment above `splitMoved` usage earlier in this function for the ceiling's own // derivation, and see the correlated-vs-window-function tradeoff explained where this query @@ -819,6 +840,7 @@ func (d *DB) Overview(f Filter) (*Overview, error) { o.CacheMiss = cacheMissM o.Uncompressed = uncompressedM o.CGLatencyMsP95, o.UpstreamMsP95 = p95cg, p95up + o.InvalidConfigRequests = invalidConfigRequests o.SafetyCost.FrozenTokens = o.FrozenTokens o.SafetyCost.RestoredTokens = o.ExpandTokens diff --git a/dash/spend_test.go b/dash/spend_test.go index 9dfc270..e44f129 100644 --- a/dash/spend_test.go +++ b/dash/spend_test.go @@ -8,13 +8,34 @@ import ( // spendEvents builds n priced requests for one tenant, split across sessions so the // quota path has whole sessions to evict. +// +// Spread one hour apart by default — comfortably distinguishable, and far enough in the +// past to satisfy every retention-age check these tests also run. MonthToDateUSD always +// asks for the CURRENT real calendar month (time.Now(), not an injectable clock — see its +// own comment on why the rollup has to be real-time), so the naive version of this that +// unconditionally went `sessions` hours into the past would occasionally split its own +// fixture across two different tenant_spend rows: measured failing, deterministically, +// the first ~10 hours of a new calendar month, because the oldest sessions land in the +// PREVIOUS month while the newest still land in this one. Clamped to the room actually +// available since local UTC midnight on the 1st, so this is exact for the ~99.9% of the +// month that isn't within `sessions` hours of the boundary, and still correct — just more +// tightly packed — for the sliver that is. func spendEvents(tenant string, sessions, turns int, usdEach float64) []*Event { - base := time.Now().Add(-time.Duration(sessions) * time.Hour).UnixMilli() + now := time.Now().UTC() + monthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC) + spacing := time.Hour + if avail := now.Sub(monthStart); avail < time.Duration(sessions)*spacing { + if spacing = avail / time.Duration(sessions+1); spacing < 10*time.Second { + spacing = 10 * time.Second // floor: keep sessions and their turns from overlapping + } + } + base := now.Add(-time.Duration(sessions) * spacing).UnixMilli() + spacingMs := spacing.Milliseconds() var evs []*Event for s := 0; s < sessions; s++ { for k := 0; k < turns; k++ { evs = append(evs, &Event{ - TS: base + int64(s)*3600_000 + int64(k)*1000, TenantID: tenant, + TS: base + int64(s)*spacingMs + int64(k)*1000, TenantID: tenant, SessionID: tenant + ":s" + string(rune('a'+s)), Model: "m", Status: 200, CostUSD: usdEach, TokenAccounting: AccountingComplete, }) diff --git a/dash/store_test.go b/dash/store_test.go index 097d35e..16d8d7f 100644 --- a/dash/store_test.go +++ b/dash/store_test.go @@ -462,6 +462,33 @@ func TestOverviewDenominatorsAndSafety(t *testing.T) { } } +// TestOverviewCountsInvalidConfigRequests pins the property the #118 incident showed was +// missing: a request forwarded uncompacted because its account's own configuration failed to +// build (proxy/tenancy.go's build() marks that row's preset "invalid" on purpose, rather than +// taking the account offline) must be visible from Overview, not only from a log line. +func TestOverviewCountsInvalidConfigRequests(t *testing.T) { + db := openTestDB(t) + ok1 := mkEvent(1000, "s1", "m", 100, 90) + ok1.Preset = "codesmart" + broken := mkEvent(2000, "s2", "m", 100, 100) + broken.Preset = "invalid" + ok2 := mkEvent(3000, "s3", "m", 100, 90) + ok2.Preset = "housellm" + if err := db.insertBatch([]*Event{ok1, broken, ok2}); err != nil { + t.Fatal(err) + } + o, err := db.Overview(Filter{}) + if err != nil { + t.Fatal(err) + } + if o.Requests != 3 { + t.Fatalf("requests = %d, want 3", o.Requests) + } + if o.InvalidConfigRequests != 1 { + t.Errorf("invalid_config_requests = %d, want 1", o.InvalidConfigRequests) + } +} + // TestNewInputRatioNeverDividesSavingsByThemselves is the guard the issue calls // non-negotiable: with no provider usage data the denominator would be `saved` // alone and the ratio would read ~100%. It must read n/a. diff --git a/dash/ui/app.js b/dash/ui/app.js index 3762ffc..8760cf0 100644 --- a/dash/ui/app.js +++ b/dash/ui/app.js @@ -1396,6 +1396,21 @@ function renderTiles(o) { const exact = (o.accounting && o.accounting.complete) || 0; const costKnown = exact > 0; + // A request forwarded with NO compaction because this account's own stored configuration + // failed to build — proxy/tenancy.go fails open on purpose rather than taking the account + // offline over a bad config row, but until this banner existed that fact lived ONLY in the + // proxy's own log. Shown above the headline, not folded into Diagnostics: the number it + // reports is "your bill was bigger than it needed to be, right now", which is exactly what + // this page exists to prevent from going unnoticed. + if (o.invalid_config_requests > 0) { + host.appendChild(el('div', { class: 'banner bad', 'data-testid': 'invalid-config' }, + el('div', {}, el('strong', {}, num(o.invalid_config_requests) + ' request' + + (o.invalid_config_requests === 1 ? '' : 's') + ' ran with NO compaction. '), + 'This account’s stored configuration failed to build, so every one of them was ' + + 'forwarded as-is instead — traffic kept working, but none of it was compacted. ' + + 'Open Settings, fix the configuration, and save it again.'))); + } + // The headline row answers the only question someone opening this page has: did it // save money, did it save tokens, and over how much traffic. Everything else is the // evidence for those three, so it sits below them in labelled groups rather than diff --git a/tenant/configmigrate.go b/tenant/configmigrate.go new file mode 100644 index 0000000..8b84844 --- /dev/null +++ b/tenant/configmigrate.go @@ -0,0 +1,245 @@ +package tenant + +// One-time recovery for the tenant configs PR #118 broke. +// +// #118 moved extract_llm's `per_output` and `cold_cache` keys to the new `extract_llm_sweep` +// component and made config.LoadBytes REFUSE either one outright — "Breaking existing configs +// is deliberate... migrated by hand" per components/offload/extract_llm.go's own comment. That +// hand migration never happened for the accounts that were already running with either key set: +// on the very next request after the new binary shipped, buildTenantConfig started failing for +// them, and the proxy's own fail-open guarantee took over — every one of their requests kept +// being forwarded, just with NO compaction applied to any of them, silently, until an operator +// happened to read the log line rather than the dashboard (see dash/overview.go for the follow-up +// fix that makes a build failure visible there instead of only in the journal). +// +// This closes that gap the way it should have shipped with #118: not by loosening the refusal +// (the refusal is right — a silently-reinterpreted `cold_cache` is "the most expensive possible +// misreading of this config", per that same comment), but by performing the EXACT mechanical +// translation #118's own migration guidance already names, in code, so it happens once, +// automatically, and is provably correct before anything is written back. +// +// A real YAML decode-modify-encode round trip, not a text-level rewrite: a first draft did this +// with regexes and every one of its bugs traced back to the same cause — a regex has no idea +// what it is looking at, so a trailing comma, a shared `trigger:` block another component also +// owns, a comment line, or a key that sorts into a different position all broke it in a +// different way, and some of those broke it SILENTLY (a config.Validate pass is not proof the +// document still says what the account meant — it is only proof the document parses and +// builds). config/form.go already decodes, edits, and re-encodes every settings-page save this +// same way (see marshalConfig's own yaml.NewEncoder(&buf); enc.SetIndent(2)), so this is not a +// new pattern in the codebase, just the first migration to use it instead of hand-rolled text +// surgery. +// +// NOTHING IS DELETED and nothing is guessed: per_output is dropped outright (the sweep "now IS +// the warm/tail pass, so there is nothing to switch off" — its presence changed nothing to begin +// with), and cold_cache's settings are carried onto a new extract_llm_sweep entry — in both the +// components map and the pipeline list, in the position config.go's own "housellm" preset uses +// — rather than discarded. Every rewritten document is round-tripped through the caller's own +// validator before it is ever written, and a tenant whose config does not match the exact shape +// this expects is left untouched and logged, never guessed at. + +import ( + "bytes" + "database/sql" + "fmt" + "log/slog" + + "gopkg.in/yaml.v3" +) + +// migrateDeprecatedExtractLLMConfig rewrites one tenant's config_yaml, moving per_output and +// cold_cache onto extract_llm_sweep exactly as #118's own migration guidance names. Returns the +// rewritten document and whether anything changed; an error means the document did not match +// the shape this can safely rewrite (extract_llm missing or not a map, cold_cache present but +// not exactly {enabled, min_tokens}, or no components/pipeline to add extract_llm_sweep to) — +// the caller's response to that is to leave the tenant alone and log it, not to guess further. +func migrateDeprecatedExtractLLMConfig(cfg string) (rewritten string, changed bool, err error) { + var doc map[string]any + if err := yaml.Unmarshal([]byte(cfg), &doc); err != nil { + return cfg, false, fmt.Errorf("could not parse as a YAML mapping: %w", err) + } + + comps, _ := doc["components"].(map[string]any) + extractLLM, _ := comps["extract_llm"].(map[string]any) + _, hasPerOutput := extractLLM["per_output"] + coldCacheRaw, hasColdCache := extractLLM["cold_cache"] + if !hasPerOutput && !hasColdCache { + return cfg, false, nil + } + if extractLLM == nil { + // hasPerOutput/hasColdCache can only be true with a non-nil extractLLM, so reaching + // here at all would itself be a bug — kept as a hard stop rather than a silent no-op. + return cfg, false, fmt.Errorf("per_output or cold_cache present but extract_llm is not a mapping") + } + + // addSweep stays false when cold_cache was already off: "cold_cache.enabled becomes the + // component's presence in the pipeline" (per extract_llm.go's own migration note) means an + // account that had already turned the sweep off needs nothing added back — just the now- + // refused key removed, same as per_output. + var sweepMinTokens int + addSweep := false + if hasColdCache { + // Decoded through a TYPED struct, not read as `any` — coldCacheRaw is a + // map[string]any at this point (the outer doc was decoded that way), and reading + // `["enabled"].(bool)` off it directly is wrong for a real config: YAML 1.1 accepts + // yes/Yes/YES/on/On/ON/y/Y as true (and their negatives as false), but decoding one of + // those into `any` yields a plain string, not a bool, so the type assertion silently + // reads it as false — exactly Bug 1's silent-loss shape again, just reached a + // different way. Re-marshaling the sub-map and decoding it through the same typed + // path config.LoadBytes itself would use fixes the bool words by construction, and + // KnownFields(true) gives the "nothing but enabled/min_tokens" check for free, so it + // replaces the extra-field bookkeeping a first draft of this had rather than adding + // to it — an account with `max_calls` or `min_idle_seconds` set is still refused, now + // because the decoder itself rejects the extra field. + raw, err := yaml.Marshal(coldCacheRaw) + if err != nil { + return cfg, false, fmt.Errorf("could not re-encode cold_cache for a typed re-read: %w", err) + } + var cc struct { + Enabled bool `yaml:"enabled"` + MinTokens *int `yaml:"min_tokens"` + } + dec := yaml.NewDecoder(bytes.NewReader(raw)) + dec.KnownFields(true) + if err := dec.Decode(&cc); err != nil { + return cfg, false, fmt.Errorf("cold_cache present but not in the enabled+min_tokens-only "+ + "shape this migration knows how to translate: %w", err) + } + if cc.Enabled { + if cc.MinTokens == nil { + return cfg, false, fmt.Errorf("cold_cache enabled but min_tokens is unset") + } + addSweep, sweepMinTokens = true, *cc.MinTokens + } + // !cc.Enabled falls through with addSweep left false: disabled, and nothing else set + // that would need translating (KnownFields already refused anything else) — drop the + // whole block, add nothing, since the sweep it would have configured never ran. + } + + if hasPerOutput { + delete(extractLLM, "per_output") + } + if hasColdCache { + delete(extractLLM, "cold_cache") + } + if addSweep { + if comps["extract_llm_sweep"] != nil { + return cfg, false, fmt.Errorf("cold_cache present but extract_llm_sweep already exists in components") + } + comps["extract_llm_sweep"] = map[string]any{"min_tokens": sweepMinTokens} + + pipeline, ok := doc["pipeline"].([]any) + if !ok { + return cfg, false, fmt.Errorf("cold_cache present but pipeline is missing or not a list") + } + idx := -1 + for i, name := range pipeline { + if s, ok := name.(string); ok && s == "extract_llm" { + idx = i + break + } + } + if idx < 0 { + return cfg, false, fmt.Errorf("cold_cache present but extract_llm does not appear in the pipeline list") + } + grown := make([]any, 0, len(pipeline)+1) + grown = append(grown, pipeline[:idx+1]...) + grown = append(grown, "extract_llm_sweep") + grown = append(grown, pipeline[idx+1:]...) + doc["pipeline"] = grown + } + + var buf bytes.Buffer + enc := yaml.NewEncoder(&buf) + enc.SetIndent(2) + if err := enc.Encode(doc); err != nil { + return cfg, false, fmt.Errorf("could not re-encode the migrated document: %w", err) + } + if err := enc.Close(); err != nil { + return cfg, false, fmt.Errorf("could not re-encode the migrated document: %w", err) + } + return buf.String(), true, nil +} + +// fixDeprecatedExtractLLMConfigs runs the migration above against every stored tenant config +// that still carries either deprecated key, once, at Open. Cheap by construction: this only +// ever matches the handful of tenants configured before #118 shipped, and once each is fixed +// (or logged as unfixable) there is nothing left for a future call to find for THAT tenant — a +// tenant this cannot safely rewrite is re-checked on every future Open, which costs one cheap +// LIKE-filtered query and is the point: it stays visible rather than being forgotten after one +// failed attempt. +// +// validate proves a rewritten document actually builds before it is ever written back — the +// same Options.Validate a caller already supplies so a user's OWN settings-page save gets +// rejected instead of silently stored broken (see Patch). Reused here rather than a second +// field: it is a parameter, not an import of the `config` package, because `config`'s own +// tests import `tenant` for a settings-form fixture, and `tenant` importing `config` back +// would be a cycle. nil — every test that opens a bare Registry, and any deployment that +// never set Options.Validate — skips this migration entirely rather than validating with +// something that would rubber-stamp anything. +// +// Best-effort per tenant and never fatal to Open: a tenant this cannot safely rewrite, OR +// whose rewrite fails to save, keeps its current (fail-open, uncompacted) behavior and is +// logged loudly, which is a strict improvement over the silent version of that same outcome +// this is replacing. A save failure for one tenant does not stop the rest from being tried. +func fixDeprecatedExtractLLMConfigs(db *sql.DB, validate func([]byte) error) error { + if validate == nil { + return nil + } + rows, err := db.Query(`SELECT id, config_yaml FROM tenants + WHERE config_yaml LIKE '%per_output%' OR config_yaml LIKE '%cold_cache%'`) + if err != nil { + return err + } + type pending struct{ id, cfg string } + var candidates []pending + for rows.Next() { + var p pending + if err := rows.Scan(&p.id, &p.cfg); err != nil { + rows.Close() + return err + } + candidates = append(candidates, p) + } + if err := rows.Err(); err != nil { + return err + } + rows.Close() + + var fixed int + for _, c := range candidates { + newCfg, changed, err := migrateDeprecatedExtractLLMConfig(c.cfg) + if err != nil { + slog.Error("tenant: could not migrate a deprecated extract_llm config; "+ + "this account keeps failing to build a compaction pipeline until fixed by hand", + "tenant", c.id, "err", err) + continue + } + if !changed { + continue + } + // The one place this whole file exists to prevent: never write a document back that + // does not provably build. The caller's validate runs the exact LoadBytes+Build path + // buildTenantConfig does in production. + if verr := validate([]byte(newCfg)); verr != nil { + slog.Error("tenant: migrated extract_llm config failed its own validation; "+ + "leaving the stored config untouched rather than writing something unproven", + "tenant", c.id, "err", verr) + continue + } + if _, err := db.Exec(`UPDATE tenants SET config_yaml = ? WHERE id = ?`, newCfg, c.id); err != nil { + // Logged, not returned: one account's write failing (a locked row, a disk error) + // must not stop every other candidate in this batch from getting its own turn — + // the same reason the two checks above use continue rather than return. + slog.Error("tenant: could not save a migrated extract_llm config; "+ + "this account keeps failing to build a compaction pipeline until fixed by hand", + "tenant", c.id, "err", err) + continue + } + fixed++ + } + if fixed > 0 { + slog.Info("tenant: recovered compaction for accounts whose config used a key #118 removed", + "accounts", fixed) + } + return nil +} diff --git a/tenant/configmigrate_test.go b/tenant/configmigrate_test.go new file mode 100644 index 0000000..c944f12 --- /dev/null +++ b/tenant/configmigrate_test.go @@ -0,0 +1,698 @@ +package tenant + +import ( + "errors" + "strings" + "testing" + + "gopkg.in/yaml.v3" +) + +// legacyExtractLLMConfig is the exact shape every affected account shared: extract_llm with +// both deprecated keys, in a flow-style pipeline list. +const legacyExtractLLMConfig = `pipeline: [format, dedup, toon, cmdfilter, searchfold, textclean, extract_llm, extract, cachesplit, toolfilter] +components: + extract: + min_tokens: 400 + extract_llm: + aggressiveness: medium + cold_cache: + enabled: true + min_tokens: 1000 + context: recent + context_messages: 2 + economic_gate: true + fire_on: pressure + llm_every_n_requests: 1 + llm_max_per_request: 8 + llm_max_per_session: 0 + min_tokens: 500 + model: + model: claude-haiku-4-5 + source: incoming + per_output: true + strategy: code + trigger: + min_request_tokens: 500 +mode: sync +` + +// asDoc decodes a migrated document for assertions that are easier to state on structure than +// on text — the whole point of moving off a text-level rewrite (see this file's package +// comment) is that "is extract_llm_sweep in the pipeline" should be answered by looking at the +// pipeline, not by grepping for a substring next to a comma. +func asDoc(t *testing.T, cfg string) map[string]any { + t.Helper() + var doc map[string]any + if err := yaml.Unmarshal([]byte(cfg), &doc); err != nil { + t.Fatalf("migrated document is not valid YAML: %v\n%s", err, cfg) + } + return doc +} + +func pipelineOf(t *testing.T, doc map[string]any) []string { + t.Helper() + raw, _ := doc["pipeline"].([]any) + out := make([]string, len(raw)) + for i, v := range raw { + s, ok := v.(string) + if !ok { + t.Fatalf("pipeline entry %d is not a string: %v", i, v) + } + out[i] = s + } + return out +} + +func componentsOf(t *testing.T, doc map[string]any) map[string]any { + t.Helper() + m, _ := doc["components"].(map[string]any) + if m == nil { + t.Fatal("components is missing or not a mapping") + } + return m +} + +func TestMigrateDeprecatedExtractLLMConfigMovesBothKeys(t *testing.T) { + out, changed, err := migrateDeprecatedExtractLLMConfig(legacyExtractLLMConfig) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + extractLLM, _ := comps["extract_llm"].(map[string]any) + if extractLLM == nil { + t.Fatal("extract_llm missing after migration") + } + if _, present := extractLLM["per_output"]; present { + t.Error("per_output still present") + } + if _, present := extractLLM["cold_cache"]; present { + t.Error("cold_cache still present") + } + sweep, _ := comps["extract_llm_sweep"].(map[string]any) + if sweep == nil { + t.Fatal("extract_llm_sweep missing") + } + if mt, ok := sweep["min_tokens"].(int); !ok || mt != 1000 { + t.Errorf("extract_llm_sweep.min_tokens = %v, want 1000", sweep["min_tokens"]) + } + pipe := pipelineOf(t, doc) + llmIdx, sweepIdx := indexOf(pipe, "extract_llm"), indexOf(pipe, "extract_llm_sweep") + if llmIdx < 0 || sweepIdx != llmIdx+1 { + t.Errorf("pipeline = %v; want extract_llm_sweep immediately after extract_llm", pipe) + } + // Everything else about the extract_llm block — the account's OWN tuning — must survive + // untouched: this is a migration, not a reset to defaults. + if mt, ok := extractLLM["min_tokens"].(int); !ok || mt != 500 { + t.Errorf("extract_llm.min_tokens = %v, want 500 (the account's own tuning)", extractLLM["min_tokens"]) + } + if extractLLM["aggressiveness"] != "medium" { + t.Errorf("extract_llm.aggressiveness = %v, want medium", extractLLM["aggressiveness"]) + } + trigger, _ := extractLLM["trigger"].(map[string]any) + if trigger == nil || trigger["min_request_tokens"] != 500 { + t.Errorf("extract_llm.trigger = %v, want {min_request_tokens: 500}", trigger) + } +} + +func indexOf(s []string, v string) int { + for i, x := range s { + if x == v { + return i + } + } + return -1 +} + +// Regression for the bug an earlier, regex-based draft of this file shipped with: a flow-style +// pipeline where extract_llm is the LAST element has no trailing comma after it, and a +// substring replace on "extract_llm," was a silent no-op there — config.Validate cannot catch +// this, because the resulting document is perfectly valid, it just never runs the sweep. A +// parse-based rewrite has no such case: the insertion point is found by comparing list +// elements, not by matching literal punctuation. +func TestMigrateDeprecatedExtractLLMConfigHandlesExtractLLMLastInPipeline(t *testing.T) { + const cfg = `pipeline: [format, dedup, toon, cmdfilter, searchfold, textclean, extract_llm] +components: + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + per_output: true + trigger: + min_request_tokens: 500 +` + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + pipe := pipelineOf(t, doc) + if got := pipe[len(pipe)-1]; got != "extract_llm_sweep" { + t.Errorf("pipeline = %v; want extract_llm_sweep as the new last element", pipe) + } + comps := componentsOf(t, doc) + if comps["extract_llm_sweep"] == nil { + t.Error("extract_llm_sweep missing from components despite the pipeline claiming it's there") + } +} + +// Regression: components.Trigger is a field shared by extract, summarize and extract_llm alike +// (each embeds it separately as yaml:"trigger"). A migration that looks for "any trigger block +// in the document" rather than "extract_llm's own trigger field" would corrupt or double-insert +// against an unrelated component. This document has trigger blocks on BOTH extract and +// extract_llm; only extract_llm's own keys may be touched. +func TestMigrateDeprecatedExtractLLMConfigDoesNotTouchAnotherComponentsTrigger(t *testing.T) { + const cfg = `pipeline: [extract_llm, extract] +components: + extract: + min_tokens: 400 + trigger: + min_request_tokens: 700 + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + trigger: + min_request_tokens: 500 +` + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + extract, _ := comps["extract"].(map[string]any) + if extract == nil { + t.Fatal("extract missing after migration") + } + trigger, _ := extract["trigger"].(map[string]any) + if trigger == nil || trigger["min_request_tokens"] != 700 { + t.Errorf("extract's own trigger = %v, want untouched {min_request_tokens: 700}", trigger) + } + if comps["extract_llm_sweep"] == nil { + t.Error("extract_llm_sweep missing") + } +} + +// Regression: extract_llm's trigger carrying more than one key (min_request_tokens is not +// necessarily first, or alone — yaml.Marshal output on this codebase's settings-form save path +// sorts keys alphabetically, and min_request_tokens sorts LAST among Trigger's fields) must not +// matter to a parse-based rewrite the way it mattered to a text anchor that only matched a +// single-key block. +func TestMigrateDeprecatedExtractLLMConfigHandlesMultiKeyTrigger(t *testing.T) { + const cfg = `pipeline: [extract_llm] +components: + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + trigger: + min_messages: 2 + min_output_tokens: 200 + min_request_tokens: 500 +` + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + extractLLM, _ := comps["extract_llm"].(map[string]any) + trigger, _ := extractLLM["trigger"].(map[string]any) + if trigger["min_messages"] != 2 || trigger["min_output_tokens"] != 200 || trigger["min_request_tokens"] != 500 { + t.Errorf("extract_llm.trigger = %v, want all three keys preserved", trigger) + } +} + +// Regression: cold_cache.enabled: false means the sweep never ran for this account — "the whole +// mechanism was off" is a real, valid state, not an unrecognized shape. It should be dropped +// with nothing added, per extract_llm.go's own migration note ("cold_cache.enabled becomes the +// component's presence in the pipeline"). +func TestMigrateDeprecatedExtractLLMConfigDropsADisabledColdCacheWithoutAddingASweep(t *testing.T) { + const cfg = `pipeline: [extract_llm] +components: + extract_llm: + cold_cache: + enabled: false + min_tokens: 1000 + per_output: true +` + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + if comps["extract_llm_sweep"] != nil { + t.Error("extract_llm_sweep added for a cold_cache that was disabled") + } + pipe := pipelineOf(t, doc) + if indexOf(pipe, "extract_llm_sweep") >= 0 { + t.Errorf("pipeline = %v; extract_llm_sweep should not be in it", pipe) + } + extractLLM, _ := comps["extract_llm"].(map[string]any) + if _, present := extractLLM["cold_cache"]; present { + t.Error("cold_cache still present") + } + if _, present := extractLLM["per_output"]; present { + t.Error("per_output still present") + } +} + +// Regression for a real blocker an independent review found in an untyped read of `enabled`: +// YAML 1.1 accepts yes/Yes/YES/on/On/ON/y/Y as true, but reading that value out of a +// map[string]any with a bare `.(bool)` assertion silently gets false (the decoder resolved it +// to a plain string, not a bool, when the target type was `any`) — an account whose sweep was +// genuinely running pre-#118 would have had it dropped with nothing added, silently, the exact +// same class of loss as the flow-pipeline bug this file's package comment already describes. +// Decoding cold_cache through a typed struct (the same path config.LoadBytes itself uses) +// fixes this by construction; this pins that it stays fixed. +func TestMigrateDeprecatedExtractLLMConfigHandlesYAML11BoolWords(t *testing.T) { + for _, word := range []string{"yes", "Yes", "YES", "on", "On", "ON", "y", "Y"} { + t.Run(word, func(t *testing.T) { + cfg := "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache:\n" + + " enabled: " + word + "\n min_tokens: 1000\n" + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + pipe := pipelineOf(t, doc) + if indexOf(pipe, "extract_llm_sweep") < 0 { + t.Errorf("enabled: %s did not add extract_llm_sweep to the pipeline (got %v)", word, pipe) + } + comps := componentsOf(t, doc) + sweep, _ := comps["extract_llm_sweep"].(map[string]any) + if sweep == nil || sweep["min_tokens"] != 1000 { + t.Errorf("enabled: %s: extract_llm_sweep = %v, want {min_tokens: 1000}", word, sweep) + } + }) + } +} + +// Companion to the above: the falsy YAML 1.1 words must still drop cleanly, matching plain +// `false` — these coincidentally already worked under the untyped read, but the typed decode +// must not regress them. +func TestMigrateDeprecatedExtractLLMConfigHandlesYAML11FalseWords(t *testing.T) { + for _, word := range []string{"no", "No", "NO", "off", "Off", "OFF", "n", "N"} { + t.Run(word, func(t *testing.T) { + cfg := "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache:\n" + + " enabled: " + word + "\n min_tokens: 1000\n" + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + if comps["extract_llm_sweep"] != nil { + t.Errorf("enabled: %s added extract_llm_sweep; a false word must drop cleanly", word) + } + }) + } +} + +// Regression: cold_cache: (YAML null) and cold_cache: {} both mean the same thing a plain +// `enabled: false` does — the typed decode's zero value is Enabled=false — so both must drop +// cleanly rather than being refused as an unrecognized shape. +func TestMigrateDeprecatedExtractLLMConfigDropsNullOrEmptyColdCache(t *testing.T) { + for name, cfg := range map[string]string{ + "null": "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache:\n per_output: true\n", + "empty": "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache: {}\n per_output: true\n", + } { + t.Run(name, func(t *testing.T) { + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + if comps["extract_llm_sweep"] != nil { + t.Error("extract_llm_sweep added for a null/empty cold_cache") + } + extractLLM, _ := comps["extract_llm"].(map[string]any) + if _, present := extractLLM["cold_cache"]; present { + t.Error("cold_cache still present") + } + }) + } +} + +// Regression: a non-bool `enabled` (e.g. hand-edited to 1 or "true") must be refused and +// logged, not silently coerced to false — such a config never built pre-#118 either (the typed +// loader rejected it the same way), so refusing here costs nothing and matches that behavior. +func TestMigrateDeprecatedExtractLLMConfigRefusesANonBoolEnabled(t *testing.T) { + for name, cfg := range map[string]string{ + "integer": "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache:\n enabled: 1\n min_tokens: 1000\n", + "string": `pipeline: [extract_llm]` + "\ncomponents:\n extract_llm:\n cold_cache:\n enabled: \"true\"\n min_tokens: 1000\n", + } { + t.Run(name, func(t *testing.T) { + _, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err == nil { + t.Fatal("expected an error for a non-bool enabled value, got nil") + } + if changed { + t.Error("changed = true on a refused document") + } + }) + } +} + +// Regression: a fourth cold_cache field (max_calls, min_idle_seconds — the two the old +// component had that the sweep has no equivalent for) must still be refused now that the extra- +// field check is KnownFields(true) on the typed decode rather than hand-rolled counting. +func TestMigrateDeprecatedExtractLLMConfigRefusesMaxCallsAndMinIdleSeconds(t *testing.T) { + for name, cfg := range map[string]string{ + "max_calls": "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache:\n enabled: true\n min_tokens: 1000\n max_calls: 3\n", + "min_idle_seconds": "pipeline: [extract_llm]\ncomponents:\n extract_llm:\n cold_cache:\n enabled: true\n min_tokens: 1000\n min_idle_seconds: 600\n", + } { + t.Run(name, func(t *testing.T) { + _, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err == nil { + t.Fatal("expected an error, got nil") + } + if changed { + t.Error("changed = true on a refused document") + } + }) + } +} + +// Regression: a "cold_cache:" substring that is not actually the key — inside a YAML comment, +// for instance — must not be mistaken for the real thing. A parse-based rewrite only ever sees +// what the document actually decodes to, so this is really a test that the migration looks at +// extractLLM["cold_cache"] and nothing textual. +func TestMigrateDeprecatedExtractLLMConfigIgnoresColdCacheMentionedInAComment(t *testing.T) { + const cfg = `# note: cold_cache: was removed upstream, revisit +pipeline: [format, extract_llm, extract] +components: + extract_llm: + min_tokens: 500 + per_output: true + trigger: + min_request_tokens: 500 +` + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true — per_output alone is still a real thing to fix") + } + doc := asDoc(t, out) + comps := componentsOf(t, doc) + extractLLM, _ := comps["extract_llm"].(map[string]any) + if _, present := extractLLM["per_output"]; present { + t.Error("per_output still present") + } + if comps["extract_llm_sweep"] != nil { + t.Error("extract_llm_sweep added despite no real cold_cache key ever being present") + } +} + +func TestMigrateDeprecatedExtractLLMConfigHandlesBlockStylePipeline(t *testing.T) { + const cfg = `cache: + head_ttl_1h: false +components: + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + per_output: true +mode: sync +pipeline: + - format + - extract_llm + - extract +` + out, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err != nil { + t.Fatal(err) + } + if !changed { + t.Fatal("changed = false, want true") + } + doc := asDoc(t, out) + pipe := pipelineOf(t, doc) + llmIdx, sweepIdx := indexOf(pipe, "extract_llm"), indexOf(pipe, "extract_llm_sweep") + if llmIdx < 0 || sweepIdx != llmIdx+1 { + t.Errorf("pipeline = %v; want extract_llm_sweep immediately after extract_llm", pipe) + } + // The unrelated cache: block, written before components: in the source, must survive. + cache, _ := doc["cache"].(map[string]any) + if cache == nil || cache["head_ttl_1h"] != false { + t.Errorf("cache block = %v, want untouched {head_ttl_1h: false}", cache) + } +} + +func TestMigrateDeprecatedExtractLLMConfigIsANoOpWithoutEitherKey(t *testing.T) { + const clean = `pipeline: [format, extract_llm, extract] +components: + extract_llm: + min_tokens: 500 + trigger: + min_request_tokens: 500 +mode: sync +` + out, changed, err := migrateDeprecatedExtractLLMConfig(clean) + if err != nil { + t.Fatal(err) + } + if changed { + t.Error("changed = true on a config with neither deprecated key") + } + if out != clean { + t.Error("output differs from input on a no-op call") + } +} + +func TestMigrateDeprecatedExtractLLMConfigRefusesAnUnrecognizedColdCacheShape(t *testing.T) { + // cold_cache present and enabled, but with a THIRD field (max_calls) beyond + // enabled/min_tokens — this migration only knows the shape every real account shared. + const weird = `components: + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + max_calls: 3 + trigger: + min_request_tokens: 500 +pipeline: [extract_llm] +` + _, changed, err := migrateDeprecatedExtractLLMConfig(weird) + if err == nil { + t.Fatal("expected an error for a cold_cache shape this migration does not recognize, got nil") + } + if changed { + t.Error("changed = true on a refused document") + } +} + +func TestMigrateDeprecatedExtractLLMConfigRefusesWhenExtractLLMSweepAlreadyExists(t *testing.T) { + // Should not happen in practice (extract_llm_sweep didn't exist when cold_cache did), but a + // hand-edited or otherwise unusual document must not silently clobber an existing entry. + const cfg = `pipeline: [extract_llm, extract_llm_sweep] +components: + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + extract_llm_sweep: + min_tokens: 2000 +` + _, changed, err := migrateDeprecatedExtractLLMConfig(cfg) + if err == nil { + t.Fatal("expected an error when extract_llm_sweep already exists, got nil") + } + if changed { + t.Error("changed = true on a refused document") + } +} + +// fixDeprecatedExtractLLMConfigs (the DB-integration half) below. + +func seedLegacyTenant(t *testing.T, r *Registry, id, cfg string) { + t.Helper() + if _, err := r.db.Exec( + `INSERT INTO tenants(id, label, email, config_yaml, created_at) VALUES (?,?,?,?,0)`, + id, id, id+"@example.test", cfg); err != nil { + t.Fatal(err) + } +} + +func TestFixDeprecatedExtractLLMConfigsRewritesAndValidates(t *testing.T) { + r, err := Open("", Options{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + seedLegacyTenant(t, r, "t1", legacyExtractLLMConfig) + + var validated string + always := func(b []byte) error { validated = string(b); return nil } + if err := fixDeprecatedExtractLLMConfigs(r.db, always); err != nil { + t.Fatal(err) + } + if validated == "" { + t.Fatal("validate was never called") + } + if strings.Contains(validated, "per_output") || strings.Contains(validated, "cold_cache") { + t.Errorf("validate was called with an unmigrated document:\n%s", validated) + } + + var stored string + if err := r.db.QueryRow(`SELECT config_yaml FROM tenants WHERE id = 't1'`).Scan(&stored); err != nil { + t.Fatal(err) + } + if stored != validated { + t.Errorf("stored config differs from the one that passed validation:\nstored: %s\nvalidated: %s", stored, validated) + } +} + +func TestFixDeprecatedExtractLLMConfigsSkipsEntirelyWithoutAValidator(t *testing.T) { + r, err := Open("", Options{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + seedLegacyTenant(t, r, "t1", legacyExtractLLMConfig) + + if err := fixDeprecatedExtractLLMConfigs(r.db, nil); err != nil { + t.Fatal(err) + } + var stored string + if err := r.db.QueryRow(`SELECT config_yaml FROM tenants WHERE id = 't1'`).Scan(&stored); err != nil { + t.Fatal(err) + } + if stored != legacyExtractLLMConfig { + t.Error("config was rewritten despite no validator being supplied") + } +} + +func TestFixDeprecatedExtractLLMConfigsNeverWritesAFailedValidation(t *testing.T) { + r, err := Open("", Options{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + seedLegacyTenant(t, r, "t1", legacyExtractLLMConfig) + + alwaysFails := func([]byte) error { return errors.New("simulated: this deployment's real validator rejected it") } + if err := fixDeprecatedExtractLLMConfigs(r.db, alwaysFails); err != nil { + t.Fatal(err) + } + var stored string + if err := r.db.QueryRow(`SELECT config_yaml FROM tenants WHERE id = 't1'`).Scan(&stored); err != nil { + t.Fatal(err) + } + if stored != legacyExtractLLMConfig { + t.Error("config was overwritten even though validation failed") + } +} + +func TestFixDeprecatedExtractLLMConfigsLeavesCleanTenantsAlone(t *testing.T) { + r, err := Open("", Options{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + const clean = `pipeline: [extract_llm] +components: + extract_llm: + min_tokens: 500 +` + seedLegacyTenant(t, r, "t1", clean) + + called := false + track := func(b []byte) error { called = true; return nil } + if err := fixDeprecatedExtractLLMConfigs(r.db, track); err != nil { + t.Fatal(err) + } + if called { + t.Error("validate was called for a tenant with neither deprecated key") + } +} + +// Regression for the isolation property between candidates: one tenant's config being +// unmigratable (a shape this doesn't recognize) or unvalidatable (the real config.Validate +// rejects the result) must not stop the OTHER candidates in the same batch from getting fixed. +func TestFixDeprecatedExtractLLMConfigsOneFailureDoesNotStarveTheOthers(t *testing.T) { + r, err := Open("", Options{}) + if err != nil { + t.Fatal(err) + } + defer r.Close() + + const unrecognizedShape = `components: + extract_llm: + cold_cache: + enabled: true + min_tokens: 1000 + max_calls: 3 +pipeline: [extract_llm] +` + seedLegacyTenant(t, r, "t_shape", unrecognizedShape) + // t_good and t_validate are identical on purpose: which one the validator happens to see + // first is not something this test can (or needs to) pin down, since SELECT ... WHERE with + // no ORDER BY makes no ordering promise. Rejecting exactly the SECOND call forces exactly + // one of these two to fail regardless of which is seen first, which is all the isolation + // property being tested needs. + seedLegacyTenant(t, r, "t_good", legacyExtractLLMConfig) + seedLegacyTenant(t, r, "t_validate", legacyExtractLLMConfig) + + calls := 0 + validate := func([]byte) error { + calls++ + if calls == 2 { + return errors.New("simulated rejection of the second candidate seen") + } + return nil + } + if err := fixDeprecatedExtractLLMConfigs(r.db, validate); err != nil { + t.Fatal(err) + } + if calls != 2 { + t.Fatalf("validate called %d times, want 2 (t_shape is refused before validate ever runs; only t_good and t_validate reach it)", calls) + } + + var shapeCfg, goodCfg, validateCfg string + r.db.QueryRow(`SELECT config_yaml FROM tenants WHERE id = 't_shape'`).Scan(&shapeCfg) + r.db.QueryRow(`SELECT config_yaml FROM tenants WHERE id = 't_good'`).Scan(&goodCfg) + r.db.QueryRow(`SELECT config_yaml FROM tenants WHERE id = 't_validate'`).Scan(&validateCfg) + + if strings.Contains(shapeCfg, "extract_llm_sweep") { + t.Error("t_shape (unrecognized shape) was migrated; it should have been refused") + } + migrated := !strings.Contains(goodCfg, "per_output") || !strings.Contains(validateCfg, "per_output") + if !migrated { + t.Error("neither t_good nor t_validate was migrated; the second candidate's failure starved the third") + } + bothMigrated := !strings.Contains(goodCfg, "per_output") && !strings.Contains(validateCfg, "per_output") + if bothMigrated { + t.Fatal("test setup did not actually force one candidate to fail validation — fix the fixture") + } +} diff --git a/tenant/tenant.go b/tenant/tenant.go index 0a9878e..69d34f3 100644 --- a/tenant/tenant.go +++ b/tenant/tenant.go @@ -31,6 +31,7 @@ import ( "encoding/hex" "errors" "fmt" + "log/slog" "os" "path/filepath" "strings" @@ -319,6 +320,13 @@ func Open(path string, o Options) (*Registry, error) { db.Close() return nil, fmt.Errorf("tenant: migrate: %w", err) } + // Schema migrations above are DDL; this is a one-time DATA fix for the config documents + // #118 broke — see configmigrate.go. Never fatal to Open: an account this cannot safely + // rewrite is logged and left running exactly as it is now (fail-open, uncompacted) rather + // than blocking every other account from starting. + if err := fixDeprecatedExtractLLMConfigs(db, o.Validate); err != nil { + slog.Error("tenant: deprecated extract_llm config recovery failed; continuing without it", "err", err) + } return &Registry{db: db, path: path, opts: o, cache: map[string]cacheEntry{}}, nil }