Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ WEBHOOK_MAX_COUNT=500
# EMBEDDING_MAX_CONCURRENT=5 (worker concurrency; default 5)
# EMBEDDING_MAX_ATTEMPTS=5 (River job attempts before a transient failure is reconciled; default 5)
# EMBEDDING_JOB_TIMEOUT_SECONDS=60 (deadline for one embedding job; independent of other enrichments)
# EMBEDDING_MAX_INPUT_BYTES=12000 (cap on the text sent to the provider, in UTF-8 bytes; over it the input is trimmed to its first N bytes. Embedding models refuse an over-long input outright, which is a terminal failure that leaves the record unsearchable, so this is what keeps long records in the index. 0 = no cap)
# EMBEDDING_BATCH_SIZE=1 (document micro-batch size; 1 disables batching; provider must support batches)
# EMBEDDING_BATCH_MAX_WAIT_MS=25 (maximum time to collect a partial batch; default 25ms)
# EMBEDDING_BATCH_MAX_IN_FLIGHT=1 (maximum concurrent provider batch requests; default 1)
Expand All @@ -152,6 +153,7 @@ WEBHOOK_MAX_COUNT=500
# TRANSLATION_DEFAULT_LANGUAGE= (optional BCP-47 fallback target, e.g. en-US; used when a tenant has no target_language of its own. Empty = per-tenant opt-in only)
# TRANSLATION_MAX_CONCURRENT=5 (worker concurrency; default 5)
# TRANSLATION_MAX_ATTEMPTS=3 (River job retries before failing; default 3)
# TRANSLATION_MAX_INPUT_BYTES=24000 (cap on the text sent to the provider, in UTF-8 bytes; over it the record is SKIPPED, not trimmed — a partial translation stored as complete is wrong — and marked failed_terminal without any provider call. An edit that brings the text back under the cap re-runs it; raising this value later does NOT clear markers already written, so records refused at the old value stay failed until their tenant clears them (POST /v1/tenants/{tenant_id}/enrichments/retry). 0 = no cap)

# Sentiment (sentiment enrichment) is optional. To enable, set both SENTIMENT_PROVIDER and SENTIMENT_MODEL; if either is unset, sentiment is disabled and no sentiment jobs run.
# Open-text feedback (value_text) is classified into a sentiment label (very_negative..very_positive, mixed) + a -1..1 polarity score. Same providers/auth model as embeddings/translation.
Expand All @@ -166,6 +168,7 @@ WEBHOOK_MAX_COUNT=500
# SENTIMENT_MODEL=<model name> (required to enable sentiment; no default; e.g. gemini-2.5-flash)
# SENTIMENT_MAX_CONCURRENT=5 (worker concurrency; default 5)
# SENTIMENT_MAX_ATTEMPTS=3 (River job retries before failing; default 3)
# SENTIMENT_MAX_INPUT_BYTES=24000 (cap on the text sent to the provider, in UTF-8 bytes; over it the input is trimmed to its first N bytes. 0 = no cap)

# Emotions (emotion enrichment) is optional and independent of sentiment. To enable, set both EMOTIONS_PROVIDER and EMOTIONS_MODEL; if either is unset, emotions is disabled and no emotion jobs run.
# Open-text feedback (value_text) is tagged with zero or more basic emotions (joy, anger, sadness, fear, surprise, disgust) — multi-label, stored as a text[] array. Same providers/auth model as sentiment.
Expand All @@ -180,6 +183,7 @@ WEBHOOK_MAX_COUNT=500
# EMOTIONS_MODEL=<model name> (required to enable emotions; no default; e.g. gemini-2.5-flash)
# EMOTIONS_MAX_CONCURRENT=5 (worker concurrency; default 5)
# EMOTIONS_MAX_ATTEMPTS=3 (River job retries before failing; default 3)
# EMOTIONS_MAX_INPUT_BYTES=24000 (cap on the text sent to the provider, in UTF-8 bytes; over it the input is trimmed to its first N bytes. 0 = no cap)

# Enrichment reconciler (hub-worker). A periodic sweep that finds eligible records still missing an
# enrichment — never enqueued, crashed mid-flight, or transiently failed past their retries — and
Expand Down
2 changes: 2 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,8 @@ Sentiment, emotions, and translation are "classify" enrichments (`text → struc

A new enrichment is then a thin per-type spec: an enum/result type in `models`, a `*_client.go` (prompt + parse + `structuredSpec`), a `*_provider.go` (config for `EnrichmentProvider`), a `*_job_args.go`, a one-line `*_worker.go` (a configured `enrichmentWorker`), a metrics adapter, a config struct, and wiring in `cmd/*`. Enrichment outputs are read-only, server-generated columns — `NULL` until enriched, cleared and re-enqueued when `value_text` changes (translation also on `language`). Writes are guarded against content supersession, so a job that read older text skips rather than overwriting a newer result; carry `tenant_id` (and, for correlation, the originating `event_id`) through job args and log lines.

**Bound what you send.** `value_text` is accepted up to 30,000 characters, so a classify hook takes its text through `trimForProvider` with that type's `*_MAX_INPUT_BYTES` cap (see `service.TextCap` for why the unit is bytes) and counts the cut in `hub_enrichment_text_trimmed_total`. Trim where a partial result is still a result — a label or a vector for the start of a record beats none — and turn the input away as a terminal `TerminalReasonLength` failure wrapping `errInputOverCap` where a partial result would be *wrong* stored as complete, which is translation's case. A pipeline that refuses must also add `observability.WorkerReasonInputOverCap` to its `allowed*WorkerReasons` in `internal/observability/names.go`: a worker-error reason that allowlist does not name is exported as `other`, and the counting doubles in the worker tests do not apply the allowlist, so assert through the real `New*Metrics`. Skipping a cross-cutting step here does not merely overspend: an over-long embedding input is a terminal provider failure, so the record stays out of the index and out of the reconciler's reach for good.

## Multi-Tenancy & Data Isolation

- Treat `tenant_id` as a security boundary, not a convenience filter. Tenant-owned data must never be read, enqueued, dispatched, cached, searched, embedded, or deleted across tenants.
Expand Down
2 changes: 2 additions & 0 deletions cmd/api/app.go
Original file line number Diff line number Diff line change
Expand Up @@ -373,6 +373,7 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
service.EmbeddingsQueueName,
cfg.Embedding.MaxAttempts,
docPrefix,
cfg.Embedding.MaxInputBytes,
embeddingMetrics,
)
messageManager.RegisterProvider(embeddingProv)
Expand All @@ -384,6 +385,7 @@ func NewApp(cfg *config.Config, db *pgxpool.Pool) (*App, error) {
service.EmbeddingsQueueName,
cfg.Embedding.MaxAttempts,
docPrefix,
cfg.Embedding.MaxInputBytes,
embeddingMetrics,
models.EmbeddingInputKindTaxonomyTranslated,
)
Expand Down
11 changes: 8 additions & 3 deletions cmd/backfill-classify/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -115,8 +115,11 @@ func run() int {
}

// nil failure recorder: this client is never Start()ed (see backfill-translations), so it
// works no jobs and records no failures; hub-worker does both.
river.AddWorker(riverWorkers, workers.NewFeedbackSentimentWorker(feedbackRecordsService, settingsService, client, nil, nil, nil))
// works no jobs and records no failures; hub-worker does both. The input cap is passed
// anyway so this registration matches what hub-worker runs.
river.AddWorker(riverWorkers, workers.NewFeedbackSentimentWorker(
feedbackRecordsService, settingsService, client, nil, nil, nil, nil,
service.TextCap{MaxBytes: cfg.Sentiment.MaxInputBytes}))

queueName = service.SentimentsQueueName
maxAttempts = classifyMaxAttempts(cfg.Sentiment.MaxAttempts)
Expand All @@ -142,7 +145,9 @@ func run() int {
return exitFailure
}

river.AddWorker(riverWorkers, workers.NewFeedbackEmotionsWorker(feedbackRecordsService, settingsService, client, nil, nil, nil))
river.AddWorker(riverWorkers, workers.NewFeedbackEmotionsWorker(
feedbackRecordsService, settingsService, client, nil, nil, nil, nil,
service.TextCap{MaxBytes: cfg.Emotions.MaxInputBytes}))

queueName = service.EmotionsQueueName
maxAttempts = classifyMaxAttempts(cfg.Emotions.MaxAttempts)
Expand Down
2 changes: 2 additions & 0 deletions cmd/backfill-embeddings/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,8 @@ func run() int {
cfg.Embedding.JobTimeout.Duration(),
nil,
nil,
cfg.Embedding.MaxInputBytes,
nil,
)
riverWorkers := river.NewWorkers()
river.AddWorker(riverWorkers, embeddingWorker)
Expand Down
4 changes: 3 additions & 1 deletion cmd/backfill-translations/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,9 @@ func run() int {
// nil metrics and nil failure recorder: this client is never Start()ed, so it works no jobs.
// The registration exists only because River validates that every configured queue has a
// worker. hub-worker is the process that runs these and owns failure bookkeeping.
river.AddWorker(riverWorkers, workers.NewFeedbackTranslationWorker(feedbackRecordsService, translationClient, nil, nil, nil))
river.AddWorker(riverWorkers, workers.NewFeedbackTranslationWorker(
feedbackRecordsService, translationClient, nil, nil, nil,
service.TextCap{MaxBytes: cfg.Translation.MaxInputBytes}))

riverClient, err := river.NewClient(riverpgxv5.New(db), &river.Config{
Queues: map[string]river.QueueConfig{
Expand Down
13 changes: 13 additions & 0 deletions cmd/worker/app.go
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,9 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) {
// The worker is where a terminal give-up is observed, so it is where the counter lives.
// nil when metrics are disabled, which the worker treats as "do not record".
FailureMetrics: failureMetrics(metrics),
// Same shape and nil-guard as FailureMetrics: the counter lives where the trimming is
// observed (the worker), and a disabled-metrics deployment simply records nothing.
TrimmedMetrics: trimmedMetrics(metrics),
// nil when the sweep is switched off or no enrichment is configured, which leaves the
// reconcile worker and its queue unregistered rather than registered-and-idle.
ReconcileSweeper: reconcileSweeperOrNil(reconcileService),
Expand Down Expand Up @@ -610,6 +613,16 @@ func reconcileMetrics(metrics *observability.Metrics) observability.EnrichmentRe
return metrics.EnrichmentReconcile
}

// trimmedMetrics is failureMetrics for the input-cap counter (ENG-3908): nil in, nil out, so a
// deployment without metrics never records a trim.
func trimmedMetrics(metrics *observability.Metrics) observability.EnrichmentTrimmedMetrics {
if metrics == nil {
return nil
}

return metrics.EnrichmentTrimmed
}

// newEnrichmentReconcileService builds the sweep, or returns nil when it should not run: the kill
// switch is off, or the deployment has no enrichment provider at all and so has nothing to sweep
// for.
Expand Down
19 changes: 19 additions & 0 deletions docs/src/content/docs/core-concepts/data-model.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,25 @@ This enables:
stored before these limits are unaffected — responses are never truncated.
</Aside>

<Aside type="note">
Records are stored whole, but each enrichment sends at most `*_MAX_INPUT_BYTES`
of a record to its provider (UTF-8 bytes, default **12,000** for embeddings and
**24,000** for translation, sentiment and emotions; `0` disables the cap). Over
the cap, translation is skipped and recorded as a terminal failure — a partial
translation stored as if it were complete is wrong — while the embeddings,
sentiment and emotions inputs are trimmed to their first N bytes. Trimming is
counted in `hub_enrichment_text_trimmed_total`.

An embedding of a trimmed record therefore describes its beginning, not the whole
answer. Raising `EMBEDDING_MAX_INPUT_BYTES` affects only embeddings created after
the change: the backfill fills missing rows and never overwrites an existing
vector, so records already embedded keep the shorter vector until their rows are
deleted and rebuilt (the `-prune-stale-models` path does that for rows left by a
previous `EMBEDDING_MODEL`). A translation refused by the cap is the easier one to
recover: an edit re-runs it, and raising the cap takes effect for those records
once their tenant clears the recorded failures.
</Aside>

<Aside type="note">
Every Hub request body is one JSON object (a JSON `null` is read as an empty
object) whose member names match the documented spelling exactly — they are
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ When a text feedback record is created, or its text changes, Hub checks whether
- **The source language** comes from the record's [`language`](/core-concepts/data-model) field. When it's absent, the provider detects the source language automatically.
- **Same-language answers are copied, not translated.** If a record's source language already matches the target (same base language and script, for example `en-GB` into `en-US`), Hub copies `value_text` verbatim into `value_text_translated` and still sets `translation_lang_key` — no provider call.
- **Edits re-translate.** Changing `value_text`, or correcting the source `language`, re-runs the translation. Emptying `value_text` clears the stored translation.
- **Over-long answers are skipped, not translated.** A record whose `value_text` is over `TRANSLATION_MAX_INPUT_BYTES` (24,000 bytes by default) is refused before any provider call and recorded as a terminal failure, so half a translation is never stored as if it were the whole text. An edit that brings the text back under the cap re-runs it; raising the cap afterwards does not clear the failures already recorded.

<Aside type="note">
Translation is **eventually consistent**. `value_text_translated` is populated
Expand Down
Loading
Loading