From 748c6c5eef8f814f3c1ea2363d7cd18c399bcd3a Mon Sep 17 00:00:00 2001 From: Wojtek Date: Fri, 22 May 2026 16:51:03 -0400 Subject: [PATCH 1/2] channel-memory: add async LLM digest worker with cost caps (#268) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an off-hot-path background worker to the channel-memory adapter that compresses verbose raw_excerpt windows into sparse topic_rollup / sequence_rollup blocks via an LLM. /digest stays LLM-free — it only reads already-generated blocks. The worker is conservative: only raw_excerpt windows are summarized (hard events, tombstones, telemetry keep their faithful deterministic blocks); results must be structured JSON citing the exact source message ids, and malformed or provenance-free output is rejected; work is cached by source ids + content hashes; each block records provider/model/version/cost in metadata_json; per-channel and per-pod daily call caps are enforced (usage tracked in a new llm_usage table) with deterministic-only fallback when disabled, over budget, or failing. Editing/deleting/forgetting a covered source dirties the rollup through shared provenance so stale summaries stop serving. The worker is disabled unless CHANNEL_MEMORY_LLM_ENABLED=true and an OpenAI-compatible CHANNEL_MEMORY_LLM_BASE_URL are configured. Tests cover verbose->sparse compression with faithful hard events, queue ordering, malformed/provenance-free rejection, cache reuse by content hash, per-pod cost-cap enforcement, deterministic fallback when disabled/failing, and edit/forget provenance invalidation (synthetic source events). Closes #268 Refs #262 --- examples/channel-memory/README.md | 36 +- examples/channel-memory/main.go | 31 +- examples/channel-memory/worker.go | 626 +++++++++++++++++++++++++ examples/channel-memory/worker_test.go | 451 ++++++++++++++++++ 4 files changed, 1140 insertions(+), 4 deletions(-) create mode 100644 examples/channel-memory/worker.go create mode 100644 examples/channel-memory/worker_test.go diff --git a/examples/channel-memory/README.md b/examples/channel-memory/README.md index ad66322..f92d07c 100644 --- a/examples/channel-memory/README.md +++ b/examples/channel-memory/README.md @@ -30,8 +30,39 @@ The deterministic path is intentionally conservative: - emits coverage-gap metadata from stored gap records - creates tombstone blocks for deleted messages without carrying deleted content -Higher-quality `topic_rollup` and `sequence_rollup` blocks belong to the async -LLM worker tracked separately. +Higher-quality `topic_rollup` and `sequence_rollup` blocks come from the async +LLM worker below. + +## Async LLM Digest Worker + +An optional background worker compresses verbose `raw_excerpt` material into +sparse `topic_rollup` / `sequence_rollup` blocks using an LLM. It is strictly +off the `/digest` hot path — `/digest` only ever reads already-generated blocks +and never calls a model. + +The worker is conservative by design: + +- only `raw_excerpt` windows are summarized; hard events, tombstones, and + telemetry blocks keep their faithful deterministic form +- every generated block requires structured JSON output citing the exact source + message ids it summarized; malformed or provenance-free results are rejected + and the deterministic blocks keep serving +- work is cached by source-message ids plus content hashes, so an unchanged + window is never re-summarized +- each block stores its provider, model, version, and cost in `metadata_json` +- conservative per-channel and per-pod daily call caps are enforced, with usage + tracked in `llm_usage`; over budget, disabled, or failing all fall back to + deterministic-only output +- editing, deleting, or forgetting a covered source dirties the rollup via + shared provenance, so stale summaries stop serving until regenerated + +It is disabled unless `CHANNEL_MEMORY_LLM_ENABLED=true` and +`CHANNEL_MEMORY_LLM_BASE_URL` (an OpenAI-compatible chat-completions endpoint) +are set. Tuning knobs: `CHANNEL_MEMORY_LLM_MODEL`, `CHANNEL_MEMORY_LLM_PROVIDER`, +`CHANNEL_MEMORY_LLM_VERSION`, `CHANNEL_MEMORY_LLM_API_KEY`, +`CHANNEL_MEMORY_LLM_WINDOW`, `CHANNEL_MEMORY_LLM_MIN_WINDOW`, +`CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY`, `CHANNEL_MEMORY_LLM_PER_POD_DAILY`, and +`CHANNEL_MEMORY_LLM_INTERVAL_SECONDS`. ## Storage @@ -47,6 +78,7 @@ The schema includes: - `derived_block_sources` - `coverage_gaps` - `processing_queue` +- `llm_usage` `source_messages` uses explicit `observed_seq`, `observed_at`, and `is_current` fields so edited messages create new rows while exact retrieval can still select diff --git a/examples/channel-memory/main.go b/examples/channel-memory/main.go index 2f38429..2803aba 100644 --- a/examples/channel-memory/main.go +++ b/examples/channel-memory/main.go @@ -225,6 +225,17 @@ func main() { } defer store.Close() + workerCfg := workerConfigFromEnv() + if workerCfg.Enabled { + client := httpLLMClientFromEnv(workerCfg) + if client == nil { + log.Printf("channel-memory digest worker enabled but CHANNEL_MEMORY_LLM_BASE_URL is unset; staying deterministic-only") + } else { + worker := newDigestWorker(store, client, workerCfg) + go worker.Run(context.Background()) + } + } + addr := strings.TrimSpace(os.Getenv("PORT")) if addr == "" { addr = defaultListenAddress @@ -358,6 +369,13 @@ func (s *channelMemoryStore) initSchema(ctx context.Context) error { updated_at TEXT NOT NULL )`, `CREATE INDEX IF NOT EXISTS idx_processing_queue_status ON processing_queue(status, updated_at)`, + `CREATE TABLE IF NOT EXISTS llm_usage ( + day TEXT NOT NULL, + scope TEXT NOT NULL, + calls INTEGER NOT NULL DEFAULT 0, + cost_usd REAL NOT NULL DEFAULT 0, + PRIMARY KEY(day, scope) + )`, } for _, stmt := range statements { if _, err := s.db.ExecContext(ctx, stmt); err != nil { @@ -648,10 +666,19 @@ func (s *channelMemoryStore) Digest(ctx context.Context, req digestRequest) (dig } rawRecent := 0 + deterministicOnly := true for _, block := range blocks { if block.Kind == "raw_excerpt" || block.Kind == "hard_event" || block.Kind == "tombstone" { rawRecent++ } + if block.Processor == digestProcessorLLM { + deterministicOnly = false + } + } + + llmCallsToday, err := s.dailyLLMCalls(ctx, s.now().UTC().Format("2006-01-02"), "pod") + if err != nil { + return digestResponse{}, err } return digestResponse{ @@ -667,8 +694,8 @@ func (s *channelMemoryStore) Digest(ctx context.Context, req digestRequest) (dig }, Blocks: blocks, Cost: digestCost{ - DeterministicOnly: true, - LLMCallsToday: 0, + DeterministicOnly: deterministicOnly, + LLMCallsToday: llmCallsToday, }, }, nil } diff --git a/examples/channel-memory/worker.go b/examples/channel-memory/worker.go new file mode 100644 index 0000000..b10b8fd --- /dev/null +++ b/examples/channel-memory/worker.go @@ -0,0 +1,626 @@ +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "os" + "sort" + "strconv" + "strings" + "time" +) + +// The async digest worker turns verbose channel material into compact sparse +// rollup blocks using an LLM, strictly off the /digest hot path. Hard events +// keep their verbatim deterministic blocks; only raw_excerpt windows are +// compressed. Every generated block carries exact source provenance, the +// worker caches by source ids + content hashes, enforces conservative daily +// call caps per channel and per pod, and falls back to deterministic-only +// output whenever it is disabled, over budget, or failing. + +const ( + digestProcessorLLM = "llm" + rollupKindTopic = "topic_rollup" + rollupKindSequence = "sequence_rollup" + + defaultWorkerWindowSize = 8 + defaultWorkerMinWindow = 3 + defaultWorkerPerChannelCalls = 24 + defaultWorkerPerPodCalls = 96 + defaultWorkerIntervalSeconds = 60 + defaultWorkerLLMTimeoutSecond = 30 +) + +type llmPromptMessage struct { + MessageID string + Author string + CreatedAt string + Content string +} + +type llmDigestPrompt struct { + Channel string + Messages []llmPromptMessage +} + +// llmDigestResult is the structured contract the worker requires from any +// model. Free-form prose without provenance is rejected. +type llmDigestResult struct { + Kind string `json:"kind"` + Text string `json:"text"` + SourceMessages []string `json:"source_messages"` + Score float64 `json:"score"` + CostUSD float64 `json:"cost_usd"` +} + +type llmDigestClient interface { + Summarize(ctx context.Context, prompt llmDigestPrompt) (llmDigestResult, error) +} + +type workerConfig struct { + Enabled bool + Provider string + Model string + Version string + WindowSize int + MinWindow int + PerChannelDailyCalls int + PerPodDailyCalls int + Interval time.Duration +} + +type digestWorker struct { + store *channelMemoryStore + client llmDigestClient + cfg workerConfig + now func() time.Time + logf func(format string, args ...any) +} + +func workerConfigFromEnv() workerConfig { + cfg := workerConfig{ + Enabled: boolEnv("CHANNEL_MEMORY_LLM_ENABLED", false), + Provider: strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_LLM_PROVIDER")), + Model: strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_LLM_MODEL")), + Version: strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_LLM_VERSION")), + WindowSize: intEnv("CHANNEL_MEMORY_LLM_WINDOW", defaultWorkerWindowSize), + MinWindow: intEnv("CHANNEL_MEMORY_LLM_MIN_WINDOW", defaultWorkerMinWindow), + PerChannelDailyCalls: intEnv("CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY", defaultWorkerPerChannelCalls), + PerPodDailyCalls: intEnv("CHANNEL_MEMORY_LLM_PER_POD_DAILY", defaultWorkerPerPodCalls), + Interval: time.Duration(intEnv("CHANNEL_MEMORY_LLM_INTERVAL_SECONDS", defaultWorkerIntervalSeconds)) * time.Second, + } + if cfg.Provider == "" { + cfg.Provider = "openai" + } + if cfg.Version == "" { + cfg.Version = cfg.Model + } + return cfg +} + +func newDigestWorker(store *channelMemoryStore, client llmDigestClient, cfg workerConfig) *digestWorker { + if cfg.WindowSize <= 0 { + cfg.WindowSize = defaultWorkerWindowSize + } + if cfg.MinWindow <= 0 { + cfg.MinWindow = defaultWorkerMinWindow + } + if cfg.MinWindow > cfg.WindowSize { + cfg.MinWindow = cfg.WindowSize + } + if cfg.Interval <= 0 { + cfg.Interval = defaultWorkerIntervalSeconds * time.Second + } + return &digestWorker{ + store: store, + client: client, + cfg: cfg, + now: store.now, + logf: log.Printf, + } +} + +// Run drives the worker on a ticker until the context is cancelled. It is a +// no-op when the worker is disabled or has no client, leaving the adapter in +// deterministic-only mode. +func (w *digestWorker) Run(ctx context.Context) { + if !w.enabled() { + w.logf("channel-memory digest worker disabled; deterministic-only digests") + return + } + ticker := time.NewTicker(w.cfg.Interval) + defer ticker.Stop() + for { + if _, err := w.processOnce(ctx); err != nil { + w.logf("channel-memory digest worker: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func (w *digestWorker) enabled() bool { + return w != nil && w.cfg.Enabled && w.client != nil +} + +// processOnce scans every channel for raw_excerpt windows that lack a fresh +// sparse rollup and compresses them, oldest window first. It returns the number +// of blocks generated. It never blocks the /digest path and degrades to a +// no-op on any failure so deterministic blocks keep serving. +func (w *digestWorker) processOnce(ctx context.Context) (int, error) { + if !w.enabled() { + return 0, nil + } + sourceKind := normalizeSourceKind("", "") + channels, err := w.store.allChannels(ctx, sourceKind) + if err != nil { + return 0, err + } + day := w.now().UTC().Format("2006-01-02") + generated := 0 + for _, channelID := range channels { + podCalls, err := w.store.dailyLLMCalls(ctx, day, "pod") + if err != nil { + return generated, err + } + if podCalls >= w.cfg.PerPodDailyCalls { + w.logf("channel-memory digest worker: per-pod daily cap %d reached", w.cfg.PerPodDailyCalls) + break + } + channelCalls, err := w.store.dailyLLMCalls(ctx, day, "channel:"+channelID) + if err != nil { + return generated, err + } + if channelCalls >= w.cfg.PerChannelDailyCalls { + continue + } + + windows, err := w.store.candidateRollupWindows(ctx, sourceKind, channelID, w.cfg.WindowSize, w.cfg.MinWindow) + if err != nil { + return generated, err + } + for _, window := range windows { + podCalls, err = w.store.dailyLLMCalls(ctx, day, "pod") + if err != nil { + return generated, err + } + if podCalls >= w.cfg.PerPodDailyCalls { + break + } + channelCalls, err = w.store.dailyLLMCalls(ctx, day, "channel:"+channelID) + if err != nil { + return generated, err + } + if channelCalls >= w.cfg.PerChannelDailyCalls { + break + } + + key := llmRollupBlockKey(sourceKind, channelID, window) + fresh, err := w.store.freshBlockExists(ctx, key) + if err != nil { + return generated, err + } + if fresh { + // Cache hit: identical source ids + content hashes already + // summarized. No LLM call. + continue + } + + result, err := w.client.Summarize(ctx, promptForWindow(channelID, window)) + if err != nil { + w.logf("channel-memory digest worker: summarize channel %s: %v", channelID, err) + _ = w.store.recordQueueFailure(ctx, sourceKind, channelID, window, err.Error()) + continue + } + if err := validateRollup(result, window); err != nil { + w.logf("channel-memory digest worker: rejected rollup for channel %s: %v", channelID, err) + _ = w.store.recordQueueFailure(ctx, sourceKind, channelID, window, err.Error()) + continue + } + meta := blockMetadata{ + Provider: w.cfg.Provider, + Model: w.cfg.Model, + Version: w.cfg.Version, + CostUSD: result.CostUSD, + } + if err := w.store.writeSparseRollup(ctx, key, sourceKind, channelID, window, result, meta, w.now()); err != nil { + return generated, err + } + if err := w.store.recordLLMUsage(ctx, day, channelID, result.CostUSD); err != nil { + return generated, err + } + generated++ + } + } + return generated, nil +} + +type rollupWindow struct { + Sources []storedSourceMessage +} + +func (win rollupWindow) ids() []string { + out := make([]string, 0, len(win.Sources)) + for _, s := range win.Sources { + out = append(out, s.MessageID) + } + return out +} + +func (win rollupWindow) idSet() map[string]string { + out := make(map[string]string, len(win.Sources)) + for _, s := range win.Sources { + out[s.MessageID] = s.ContentHash + } + return out +} + +func promptForWindow(channelID string, window rollupWindow) llmDigestPrompt { + msgs := make([]llmPromptMessage, 0, len(window.Sources)) + for _, s := range window.Sources { + msgs = append(msgs, llmPromptMessage{ + MessageID: s.MessageID, + Author: firstNonEmpty(s.AuthorName, s.AuthorID, "unknown"), + CreatedAt: s.CreatedAt, + Content: s.Content, + }) + } + return llmDigestPrompt{Channel: channelID, Messages: msgs} +} + +func validateRollup(result llmDigestResult, window rollupWindow) error { + switch result.Kind { + case rollupKindTopic, rollupKindSequence: + default: + return fmt.Errorf("unexpected rollup kind %q", result.Kind) + } + if strings.TrimSpace(result.Text) == "" { + return errors.New("empty rollup text") + } + cited := trimNonEmpty(result.SourceMessages) + if len(cited) == 0 { + return errors.New("rollup is provenance-free (no source_messages)") + } + allowed := window.idSet() + for _, id := range cited { + if _, ok := allowed[id]; !ok { + return fmt.Errorf("rollup cites source %q outside its window", id) + } + } + return nil +} + +func llmRollupBlockKey(sourceKind, channelID string, window rollupWindow) string { + parts := make([]string, 0, len(window.Sources)) + for _, s := range window.Sources { + parts = append(parts, s.MessageID+"@"+s.ContentHash) + } + sort.Strings(parts) + sum := sha256.Sum256([]byte(strings.Join(parts, "\x00"))) + return strings.Join([]string{"llm", sourceKind, channelID, hex.EncodeToString(sum[:])}, ":") +} + +type blockMetadata struct { + Provider string `json:"provider"` + Model string `json:"model"` + Version string `json:"version"` + CostUSD float64 `json:"cost_usd"` +} + +// candidateRollupWindows groups the channel's current, non-deleted, +// non-forgotten raw_excerpt messages (the verbose, low-signal material) into +// ordered windows. Hard events and telemetry noise are excluded so they keep +// their faithful deterministic blocks. +func (s *channelMemoryStore) candidateRollupWindows(ctx context.Context, sourceKind, channelID string, windowSize, minWindow int) ([]rollupWindow, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, source_kind, channel_id, message_id, content_hash, author_id, author_name, + created_at, edited_at, deleted, content, service, surface, guild_id, visibility_scope, + observed_seq, observed_at, is_current + FROM source_messages + WHERE source_kind = ? AND channel_id = ? AND is_current = 1 AND deleted = 0 AND forgotten_at = '' + ORDER BY created_at, observed_seq`, + sourceKind, channelID, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + eligible := make([]storedSourceMessage, 0) + for rows.Next() { + record, err := scanStoredSource(rows) + if err != nil { + return nil, err + } + if isTelemetryNoise(record.Content) { + continue + } + if kind, _, _ := classifySourceContent(record.Content); kind != "raw_excerpt" { + continue + } + eligible = append(eligible, record) + } + if err := rows.Err(); err != nil { + return nil, err + } + + windows := make([]rollupWindow, 0) + for start := 0; start < len(eligible); start += windowSize { + end := start + windowSize + if end > len(eligible) { + end = len(eligible) + } + chunk := eligible[start:end] + if len(chunk) < minWindow { + break + } + windows = append(windows, rollupWindow{Sources: append([]storedSourceMessage(nil), chunk...)}) + } + return windows, nil +} + +func (s *channelMemoryStore) freshBlockExists(ctx context.Context, blockKey string) (bool, error) { + var one int + err := s.db.QueryRowContext(ctx, + `SELECT 1 FROM derived_blocks WHERE block_key = ? AND stale = 0 AND dirty = 0`, + blockKey, + ).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + return true, nil +} + +// writeSparseRollup persists the LLM rollup block with full source provenance +// and stales the per-message raw_excerpt blocks it subsumes, so /digest serves +// the compact rollup instead of the verbose lines. +func (s *channelMemoryStore) writeSparseRollup(ctx context.Context, blockKey, sourceKind, channelID string, window rollupWindow, result llmDigestResult, meta blockMetadata, now time.Time) error { + if len(window.Sources) == 0 { + return errors.New("empty rollup window") + } + metaJSON, err := json.Marshal(meta) + if err != nil { + return err + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer rollbackUnlessCommitted(tx) + + from := window.Sources[0].CreatedAt + to := window.Sources[len(window.Sources)-1].CreatedAt + generatedAt := now.UTC().Format(time.RFC3339) + score := result.Score + if score <= 0 { + score = 0.5 + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO derived_blocks( + block_key, kind, event_type, text, source_channel, + source_window_from, source_window_to, sparse, score, generated_at, stale, dirty, processor, metadata_json + ) VALUES (?, ?, '', ?, ?, ?, ?, 1, ?, ?, 0, 0, ?, ?) + ON CONFLICT(block_key) DO UPDATE SET + kind = excluded.kind, + text = excluded.text, + source_window_from = excluded.source_window_from, + source_window_to = excluded.source_window_to, + score = excluded.score, + generated_at = excluded.generated_at, + stale = 0, + dirty = 0, + processor = excluded.processor, + metadata_json = excluded.metadata_json`, + blockKey, result.Kind, strings.TrimSpace(result.Text), channelID, + from, to, score, generatedAt, digestProcessorLLM, string(metaJSON), + ); err != nil { + return err + } + var blockID int64 + if err := tx.QueryRowContext(ctx, `SELECT id FROM derived_blocks WHERE block_key = ?`, blockKey).Scan(&blockID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM derived_block_sources WHERE block_id = ?`, blockID); err != nil { + return err + } + for _, src := range window.Sources { + if _, err := tx.ExecContext(ctx, ` + INSERT OR IGNORE INTO derived_block_sources(block_id, source_kind, channel_id, message_id, content_hash) + VALUES (?, ?, ?, ?, ?)`, + blockID, src.SourceKind, src.ChannelID, src.MessageID, src.ContentHash, + ); err != nil { + return err + } + // Stale the verbose deterministic raw_excerpt block this rollup + // subsumes. Hard events / tombstones / telemetry blocks are untouched. + if _, err := tx.ExecContext(ctx, ` + UPDATE derived_blocks + SET stale = 1 + WHERE processor = 'deterministic' AND kind = 'raw_excerpt' AND id IN ( + SELECT block_id FROM derived_block_sources + WHERE source_kind = ? AND channel_id = ? AND message_id = ? AND content_hash = ? + ) AND id <> ?`, + src.SourceKind, src.ChannelID, src.MessageID, src.ContentHash, blockID, + ); err != nil { + return err + } + } + return tx.Commit() +} + +func (s *channelMemoryStore) dailyLLMCalls(ctx context.Context, day, scope string) (int, error) { + var calls int + err := s.db.QueryRowContext(ctx, + `SELECT calls FROM llm_usage WHERE day = ? AND scope = ?`, + day, scope, + ).Scan(&calls) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + if err != nil { + return 0, err + } + return calls, nil +} + +func (s *channelMemoryStore) recordLLMUsage(ctx context.Context, day, channelID string, costUSD float64) error { + for _, scope := range []string{"pod", "channel:" + channelID} { + if _, err := s.db.ExecContext(ctx, ` + INSERT INTO llm_usage(day, scope, calls, cost_usd) + VALUES (?, ?, 1, ?) + ON CONFLICT(day, scope) DO UPDATE SET + calls = calls + 1, + cost_usd = cost_usd + excluded.cost_usd`, + day, scope, costUSD, + ); err != nil { + return err + } + } + return nil +} + +func (s *channelMemoryStore) recordQueueFailure(ctx context.Context, sourceKind, channelID string, window rollupWindow, message string) error { + now := s.now().UTC().Format(time.RFC3339) + for _, src := range window.Sources { + if _, err := s.db.ExecContext(ctx, ` + UPDATE processing_queue + SET attempts = attempts + 1, last_error = ?, updated_at = ? + WHERE source_kind = ? AND channel_id = ? AND message_id = ? AND content_hash = ?`, + truncateError(message), now, src.SourceKind, channelID, src.MessageID, src.ContentHash, + ); err != nil { + return err + } + } + return nil +} + +func truncateError(message string) string { + message = strings.TrimSpace(message) + if len(message) > 500 { + return message[:500] + } + return message +} + +// httpLLMClient calls an OpenAI-compatible chat completion endpoint and expects +// a JSON object matching llmDigestResult. It is intentionally minimal; richer +// providers can implement llmDigestClient directly. +type httpLLMClient struct { + baseURL string + apiKey string + model string + client *http.Client +} + +func httpLLMClientFromEnv(cfg workerConfig) llmDigestClient { + baseURL := strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_LLM_BASE_URL")) + if baseURL == "" { + return nil + } + return &httpLLMClient{ + baseURL: strings.TrimRight(baseURL, "/"), + apiKey: strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_LLM_API_KEY")), + model: cfg.Model, + client: &http.Client{Timeout: defaultWorkerLLMTimeoutSecond * time.Second}, + } +} + +func (c *httpLLMClient) Summarize(ctx context.Context, prompt llmDigestPrompt) (llmDigestResult, error) { + var transcript strings.Builder + for _, m := range prompt.Messages { + fmt.Fprintf(&transcript, "[%s] (id=%s) %s: %s\n", m.CreatedAt, m.MessageID, m.Author, m.Content) + } + system := "You compress Discord channel transcripts into one compact digest block. " + + "Respond ONLY with a JSON object: {\"kind\":\"topic_rollup\"|\"sequence_rollup\",\"text\":string,\"source_messages\":[message ids you summarized],\"score\":0..1}. " + + "source_messages MUST list the exact message ids you used. Do not invent ids." + payload := map[string]any{ + "model": c.model, + "messages": []map[string]string{ + {"role": "system", "content": system}, + {"role": "user", "content": "Channel " + prompt.Channel + " transcript:\n" + transcript.String()}, + }, + "response_format": map[string]string{"type": "json_object"}, + "temperature": "0", + } + body, err := json.Marshal(payload) + if err != nil { + return llmDigestResult{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/chat/completions", bytes.NewReader(body)) + if err != nil { + return llmDigestResult{}, err + } + req.Header.Set("Content-Type", "application/json") + if c.apiKey != "" { + req.Header.Set("Authorization", "Bearer "+c.apiKey) + } + resp, err := c.client.Do(req) + if err != nil { + return llmDigestResult{}, err + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if err != nil { + return llmDigestResult{}, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return llmDigestResult{}, fmt.Errorf("llm endpoint returned %s: %s", resp.Status, strings.TrimSpace(string(raw))) + } + var completion struct { + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + } `json:"choices"` + } + if err := json.Unmarshal(raw, &completion); err != nil { + return llmDigestResult{}, fmt.Errorf("decode completion: %w", err) + } + if len(completion.Choices) == 0 { + return llmDigestResult{}, errors.New("llm endpoint returned no choices") + } + var result llmDigestResult + if err := json.Unmarshal([]byte(completion.Choices[0].Message.Content), &result); err != nil { + return llmDigestResult{}, fmt.Errorf("decode rollup json: %w", err) + } + return result, nil +} + +func boolEnv(name string, fallback bool) bool { + raw := strings.TrimSpace(os.Getenv(name)) + if raw == "" { + return fallback + } + value, err := strconv.ParseBool(raw) + if err != nil { + return fallback + } + return value +} + +func intEnv(name string, fallback int) int { + raw := strings.TrimSpace(os.Getenv(name)) + if raw == "" { + return fallback + } + value, err := strconv.Atoi(raw) + if err != nil || value <= 0 { + return fallback + } + return value +} diff --git a/examples/channel-memory/worker_test.go b/examples/channel-memory/worker_test.go new file mode 100644 index 0000000..14835b4 --- /dev/null +++ b/examples/channel-memory/worker_test.go @@ -0,0 +1,451 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" +) + +// fakeLLMClient records prompts and returns scripted results so worker behavior +// is deterministic in tests. +type fakeLLMClient struct { + mu sync.Mutex + prompts []llmDigestPrompt + respond func(llmDigestPrompt) (llmDigestResult, error) +} + +func (f *fakeLLMClient) Summarize(_ context.Context, prompt llmDigestPrompt) (llmDigestResult, error) { + f.mu.Lock() + f.prompts = append(f.prompts, prompt) + f.mu.Unlock() + return f.respond(prompt) +} + +func (f *fakeLLMClient) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.prompts) +} + +// citeAllTopicRollup is the happy-path fake: a topic_rollup citing every message +// in the window. +func citeAllTopicRollup(prompt llmDigestPrompt) (llmDigestResult, error) { + ids := make([]string, 0, len(prompt.Messages)) + for _, m := range prompt.Messages { + ids = append(ids, m.MessageID) + } + return llmDigestResult{ + Kind: rollupKindTopic, + Text: fmt.Sprintf("Rollup of %d messages in %s.", len(ids), prompt.Channel), + SourceMessages: ids, + Score: 0.7, + CostUSD: 0.002, + }, nil +} + +func testWorker(store *channelMemoryStore, client llmDigestClient, mutate func(*workerConfig)) *digestWorker { + cfg := workerConfig{ + Enabled: true, + Provider: "openai", + Model: "gpt-4o-mini", + Version: "gpt-4o-mini", + WindowSize: 3, + MinWindow: 3, + PerChannelDailyCalls: 100, + PerPodDailyCalls: 100, + Interval: time.Minute, + } + if mutate != nil { + mutate(&cfg) + } + return newDigestWorker(store, client, cfg) +} + +func ingestRaw(t *testing.T, store *channelMemoryStore, channel, id, content string, created time.Time, hash string) { + t.Helper() + _, err := store.Ingest(context.Background(), ingestRequest{ + ChannelID: channel, + Message: ingestMessage{ + ID: id, + AuthorName: "member-" + id, + CreatedAt: created.UTC().Format(time.RFC3339), + Content: content, + ContentHash: hash, + }, + }) + if err != nil { + t.Fatalf("ingest %s: %v", id, err) + } +} + +func digestFor(t *testing.T, store *channelMemoryStore, channel string) digestResponse { + t.Helper() + resp, err := store.Digest(context.Background(), digestRequest{ChannelIDs: []string{channel}, Since: "24h"}) + if err != nil { + t.Fatalf("digest: %v", err) + } + return resp +} + +func llmBlockState(t *testing.T, store *channelMemoryStore) (found bool, kind string, sparse, dirty, stale bool) { + t.Helper() + var k string + var sp, dr, st int + // Any error (including sql.ErrNoRows) means there is no LLM block. + if err := store.db.QueryRowContext(context.Background(), + `SELECT kind, sparse, dirty, stale FROM derived_blocks WHERE processor = 'llm' ORDER BY id LIMIT 1`, + ).Scan(&k, &sp, &dr, &st); err != nil { + return false, "", false, false, false + } + return true, k, sp != 0, dr != 0, st != 0 +} + +// TestDigestWorkerCompressesVerboseToSparseAndKeepsHardEvents proves the core +// #262 win: verbose raw_excerpt material collapses into one sparse, fully +// provenanced rollup while hard events stay verbatim, and the digest flips out +// of deterministic-only mode. +func TestDigestWorkerCompressesVerboseToSparseAndKeepsHardEvents(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + + // One hard event that must be preserved verbatim. + _, err := store.Ingest(context.Background(), ingestRequest{ + ChannelID: "chan-1", + Message: ingestMessage{ + ID: "100", AuthorName: "lead", CreatedAt: base.Format(time.RFC3339), + Content: "[PROPOSED] signal-100 BUY ACME", ContentHash: "sha256:hard-100", + }, + }) + if err != nil { + t.Fatalf("ingest hard event: %v", err) + } + // Three verbose chatter messages (classify as raw_excerpt). + ingestRaw(t, store, "chan-1", "101", "anyone grabbing lunch later today", base.Add(time.Minute), "sha256:r101") + ingestRaw(t, store, "chan-1", "102", "the office coffee machine is broken again", base.Add(2*time.Minute), "sha256:r102") + ingestRaw(t, store, "chan-1", "103", "weather looks nice for the weekend", base.Add(3*time.Minute), "sha256:r103") + + client := &fakeLLMClient{respond: citeAllTopicRollup} + worker := testWorker(store, client, nil) + + generated, err := worker.processOnce(context.Background()) + if err != nil { + t.Fatalf("processOnce: %v", err) + } + if generated != 1 { + t.Fatalf("expected 1 sparse block generated, got %d", generated) + } + if client.callCount() != 1 { + t.Fatalf("expected exactly 1 LLM call, got %d", client.callCount()) + } + + resp := digestFor(t, store, "chan-1") + if resp.Cost.DeterministicOnly { + t.Fatalf("expected deterministic_only=false once an LLM block exists: %+v", resp.Cost) + } + if resp.Cost.LLMCallsToday != 1 { + t.Fatalf("expected llm_calls_today=1, got %d", resp.Cost.LLMCallsToday) + } + + var hardEvents, sparseRollups int + var sparse digestBlock + for _, b := range resp.Blocks { + switch { + case b.Kind == "hard_event": + hardEvents++ + if b.Text != "[18:00] lead: [PROPOSED] signal-100 BUY ACME" { + t.Fatalf("hard event text not preserved verbatim: %q", b.Text) + } + case b.Processor == digestProcessorLLM: + sparseRollups++ + sparse = b + case b.Kind == "raw_excerpt": + t.Fatalf("verbose raw_excerpt block %v should have been compressed away", b.SourceMessages) + } + } + if hardEvents != 1 { + t.Fatalf("expected hard event preserved, got %d", hardEvents) + } + if sparseRollups != 1 { + t.Fatalf("expected 1 sparse rollup, got %d", sparseRollups) + } + if !sparse.Sparse || sparse.Kind != rollupKindTopic { + t.Fatalf("unexpected sparse block: %+v", sparse) + } + // Provenance: rollup must cite exactly the three verbose messages. + want := map[string]bool{"101": true, "102": true, "103": true} + if len(sparse.SourceMessages) != 3 { + t.Fatalf("expected 3 source messages on rollup, got %v", sparse.SourceMessages) + } + for _, id := range sparse.SourceMessages { + if !want[id] { + t.Fatalf("rollup cites unexpected source %q", id) + } + } + // Compression: 4 source messages -> 2 served blocks. + if len(resp.Blocks) != 2 { + t.Fatalf("expected 2 served blocks (hard event + rollup), got %d: %+v", len(resp.Blocks), resp.Blocks) + } +} + +// TestDigestWorkerRejectsMalformedAndProvenanceFreeOutput proves invalid model +// output never becomes a block and the adapter stays deterministic. +func TestDigestWorkerRejectsMalformedAndProvenanceFreeOutput(t *testing.T) { + cases := []struct { + name string + respond func(llmDigestPrompt) (llmDigestResult, error) + }{ + {"bad kind", func(p llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{Kind: "freeform", Text: "stuff", SourceMessages: []string{p.Messages[0].MessageID}}, nil + }}, + {"empty text", func(p llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{Kind: rollupKindTopic, Text: " ", SourceMessages: []string{p.Messages[0].MessageID}}, nil + }}, + {"provenance free", func(p llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{Kind: rollupKindTopic, Text: "summary", SourceMessages: nil}, nil + }}, + {"hallucinated source", func(p llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{Kind: rollupKindTopic, Text: "summary", SourceMessages: []string{"999999"}}, nil + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + ingestRaw(t, store, "chan-1", "201", "chat one about nothing", base, "sha256:r201") + ingestRaw(t, store, "chan-1", "202", "chat two about nothing", base.Add(time.Minute), "sha256:r202") + ingestRaw(t, store, "chan-1", "203", "chat three about nothing", base.Add(2*time.Minute), "sha256:r203") + + client := &fakeLLMClient{respond: tc.respond} + worker := testWorker(store, client, nil) + generated, err := worker.processOnce(context.Background()) + if err != nil { + t.Fatalf("processOnce: %v", err) + } + if generated != 0 { + t.Fatalf("expected 0 blocks generated for %s, got %d", tc.name, generated) + } + if found, _, _, _, _ := llmBlockState(t, store); found { + t.Fatalf("malformed output (%s) must not produce an llm block", tc.name) + } + resp := digestFor(t, store, "chan-1") + if !resp.Cost.DeterministicOnly { + t.Fatalf("expected deterministic-only fallback after rejection (%s)", tc.name) + } + // Deterministic raw_excerpt blocks remain served. + rawCount := 0 + for _, b := range resp.Blocks { + if b.Kind == "raw_excerpt" { + rawCount++ + } + } + if rawCount != 3 { + t.Fatalf("expected 3 deterministic raw_excerpt blocks to remain (%s), got %d", tc.name, rawCount) + } + // Failure is recorded against the queue. + var attempts int + if err := store.db.QueryRowContext(context.Background(), + `SELECT COALESCE(MAX(attempts),0) FROM processing_queue WHERE channel_id = 'chan-1'`, + ).Scan(&attempts); err != nil { + t.Fatalf("queue attempts: %v", err) + } + if attempts < 1 { + t.Fatalf("expected queue failure recorded for %s", tc.name) + } + }) + } +} + +// TestDigestWorkerCachesByContentHash proves identical windows are not +// re-summarized. +func TestDigestWorkerCachesByContentHash(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + ingestRaw(t, store, "chan-1", "301", "first idle message", base, "sha256:r301") + ingestRaw(t, store, "chan-1", "302", "second idle message", base.Add(time.Minute), "sha256:r302") + ingestRaw(t, store, "chan-1", "303", "third idle message", base.Add(2*time.Minute), "sha256:r303") + + client := &fakeLLMClient{respond: citeAllTopicRollup} + worker := testWorker(store, client, nil) + + if _, err := worker.processOnce(context.Background()); err != nil { + t.Fatalf("processOnce #1: %v", err) + } + if client.callCount() != 1 { + t.Fatalf("expected 1 call after first pass, got %d", client.callCount()) + } + // Second pass over identical source ids + content hashes: cache hit, no call. + if _, err := worker.processOnce(context.Background()); err != nil { + t.Fatalf("processOnce #2: %v", err) + } + if client.callCount() != 1 { + t.Fatalf("expected cache hit (still 1 call), got %d", client.callCount()) + } +} + +// TestDigestWorkerEnforcesDailyCostCaps proves the per-pod daily cap stops LLM +// calls and the remaining windows fall back to deterministic. +func TestDigestWorkerEnforcesDailyCostCaps(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + // Two full windows in one channel (6 raw messages, window size 3). + for i := 0; i < 6; i++ { + id := fmt.Sprintf("4%02d", i) + ingestRaw(t, store, "chan-1", id, "idle chatter "+id, base.Add(time.Duration(i)*time.Minute), "sha256:r"+id) + } + + client := &fakeLLMClient{respond: citeAllTopicRollup} + worker := testWorker(store, client, func(c *workerConfig) { + c.PerPodDailyCalls = 1 + c.PerChannelDailyCalls = 100 + }) + + generated, err := worker.processOnce(context.Background()) + if err != nil { + t.Fatalf("processOnce: %v", err) + } + if generated != 1 { + t.Fatalf("expected pod cap to allow exactly 1 block, got %d", generated) + } + if client.callCount() != 1 { + t.Fatalf("expected pod cap to allow exactly 1 LLM call, got %d", client.callCount()) + } + // Queue ordering: the single allowed call must summarize the OLDEST window + // (messages 400,401,402), not a later one. + if len(client.prompts) != 1 { + t.Fatalf("expected 1 recorded prompt, got %d", len(client.prompts)) + } + gotIDs := make([]string, 0, len(client.prompts[0].Messages)) + for _, m := range client.prompts[0].Messages { + gotIDs = append(gotIDs, m.MessageID) + } + wantIDs := []string{"400", "401", "402"} + if fmt.Sprint(gotIDs) != fmt.Sprint(wantIDs) { + t.Fatalf("expected oldest window %v summarized first, got %v", wantIDs, gotIDs) + } + // Re-running stays capped (usage persists for the day). + if _, err := worker.processOnce(context.Background()); err != nil { + t.Fatalf("processOnce #2: %v", err) + } + if client.callCount() != 1 { + t.Fatalf("expected still 1 call after cap, got %d", client.callCount()) + } +} + +// TestDigestWorkerDeterministicFallbackWhenDisabledOrFailing proves the worker +// is inert when disabled and harmless when the model errors. +func TestDigestWorkerDeterministicFallbackWhenDisabledOrFailing(t *testing.T) { + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + + t.Run("disabled", func(t *testing.T) { + store := newTestStore(t) + defer store.Close() + ingestRaw(t, store, "chan-1", "501", "idle one", base, "sha256:r501") + ingestRaw(t, store, "chan-1", "502", "idle two", base.Add(time.Minute), "sha256:r502") + ingestRaw(t, store, "chan-1", "503", "idle three", base.Add(2*time.Minute), "sha256:r503") + client := &fakeLLMClient{respond: citeAllTopicRollup} + worker := testWorker(store, client, func(c *workerConfig) { c.Enabled = false }) + generated, err := worker.processOnce(context.Background()) + if err != nil { + t.Fatalf("processOnce: %v", err) + } + if generated != 0 || client.callCount() != 0 { + t.Fatalf("disabled worker must not call the model: generated=%d calls=%d", generated, client.callCount()) + } + if !digestFor(t, store, "chan-1").Cost.DeterministicOnly { + t.Fatal("disabled worker must leave deterministic-only digests") + } + }) + + t.Run("failing", func(t *testing.T) { + store := newTestStore(t) + defer store.Close() + ingestRaw(t, store, "chan-1", "511", "idle one", base, "sha256:r511") + ingestRaw(t, store, "chan-1", "512", "idle two", base.Add(time.Minute), "sha256:r512") + ingestRaw(t, store, "chan-1", "513", "idle three", base.Add(2*time.Minute), "sha256:r513") + client := &fakeLLMClient{respond: func(llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{}, errors.New("upstream 503") + }} + worker := testWorker(store, client, nil) + generated, err := worker.processOnce(context.Background()) + if err != nil { + t.Fatalf("processOnce should swallow model errors: %v", err) + } + if generated != 0 { + t.Fatalf("failing model must not produce blocks, got %d", generated) + } + resp := digestFor(t, store, "chan-1") + if !resp.Cost.DeterministicOnly { + t.Fatal("failing worker must leave deterministic-only digests") + } + rawCount := 0 + for _, b := range resp.Blocks { + if b.Kind == "raw_excerpt" { + rawCount++ + } + } + if rawCount != 3 { + t.Fatalf("expected 3 deterministic blocks to keep serving, got %d", rawCount) + } + }) +} + +// TestDigestWorkerEditAndForgetMarkRollupStaleOrDirty proves provenance-driven +// invalidation: editing or forgetting a covered source dirties the rollup so it +// stops serving until regenerated. Uses synthetic source events because +// claw-wall only observes first sightings. +func TestDigestWorkerEditAndForgetMarkRollupStaleOrDirty(t *testing.T) { + t.Run("edit dirties rollup", func(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + ingestRaw(t, store, "chan-1", "601", "idle one", base, "sha256:r601-v1") + ingestRaw(t, store, "chan-1", "602", "idle two", base.Add(time.Minute), "sha256:r602") + ingestRaw(t, store, "chan-1", "603", "idle three", base.Add(2*time.Minute), "sha256:r603") + worker := testWorker(store, &fakeLLMClient{respond: citeAllTopicRollup}, nil) + if _, err := worker.processOnce(context.Background()); err != nil { + t.Fatalf("processOnce: %v", err) + } + if found, _, _, dirty, _ := llmBlockState(t, store); !found || dirty { + t.Fatalf("expected a fresh non-dirty rollup, found=%v dirty=%v", found, dirty) + } + // Synthetic edit of message 601: new content hash for same id. + ingestRaw(t, store, "chan-1", "601", "idle one (edited)", base, "sha256:r601-v2") + found, _, _, dirty, _ := llmBlockState(t, store) + if !found || !dirty { + t.Fatalf("expected rollup dirtied after edit, found=%v dirty=%v", found, dirty) + } + // Dirty blocks are not served. + for _, b := range digestFor(t, store, "chan-1").Blocks { + if b.Processor == digestProcessorLLM { + t.Fatal("dirty rollup must not be served") + } + } + }) + + t.Run("forget dirties rollup", func(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + ingestRaw(t, store, "chan-1", "701", "idle one", base, "sha256:r701") + ingestRaw(t, store, "chan-1", "702", "idle two", base.Add(time.Minute), "sha256:r702") + ingestRaw(t, store, "chan-1", "703", "idle three", base.Add(2*time.Minute), "sha256:r703") + worker := testWorker(store, &fakeLLMClient{respond: citeAllTopicRollup}, nil) + if _, err := worker.processOnce(context.Background()); err != nil { + t.Fatalf("processOnce: %v", err) + } + if _, err := store.Forget(context.Background(), forgetRequest{ChannelID: "chan-1", MessageIDs: []string{"702"}, Reason: "pii"}); err != nil { + t.Fatalf("forget: %v", err) + } + found, _, _, dirty, _ := llmBlockState(t, store) + if !found || !dirty { + t.Fatalf("expected rollup dirtied after forget, found=%v dirty=%v", found, dirty) + } + }) +} From a338f8fbdf699fd27e2cd7586a0befbb46daa1d0 Mon Sep 17 00:00:00 2001 From: Wojtek Date: Sat, 23 May 2026 13:37:21 -0400 Subject: [PATCH 2/2] channel-memory: harden async digest worker caps --- examples/channel-memory/README.md | 5 +- examples/channel-memory/main.go | 93 ++++++++++++- examples/channel-memory/worker.go | 176 +++++++++++++++++++++---- examples/channel-memory/worker_test.go | 109 +++++++++++++++ 4 files changed, 349 insertions(+), 34 deletions(-) diff --git a/examples/channel-memory/README.md b/examples/channel-memory/README.md index f92d07c..ecaefec 100644 --- a/examples/channel-memory/README.md +++ b/examples/channel-memory/README.md @@ -61,7 +61,10 @@ It is disabled unless `CHANNEL_MEMORY_LLM_ENABLED=true` and are set. Tuning knobs: `CHANNEL_MEMORY_LLM_MODEL`, `CHANNEL_MEMORY_LLM_PROVIDER`, `CHANNEL_MEMORY_LLM_VERSION`, `CHANNEL_MEMORY_LLM_API_KEY`, `CHANNEL_MEMORY_LLM_WINDOW`, `CHANNEL_MEMORY_LLM_MIN_WINDOW`, -`CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY`, `CHANNEL_MEMORY_LLM_PER_POD_DAILY`, and +`CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY`, `CHANNEL_MEMORY_LLM_PER_POD_DAILY`, +`CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY_USD`, +`CHANNEL_MEMORY_LLM_PER_POD_DAILY_USD`, +`CHANNEL_MEMORY_LLM_COST_PER_CALL_USD`, and `CHANNEL_MEMORY_LLM_INTERVAL_SECONDS`. ## Storage diff --git a/examples/channel-memory/main.go b/examples/channel-memory/main.go index 2803aba..e7fd986 100644 --- a/examples/channel-memory/main.go +++ b/examples/channel-memory/main.go @@ -229,7 +229,7 @@ func main() { if workerCfg.Enabled { client := httpLLMClientFromEnv(workerCfg) if client == nil { - log.Printf("channel-memory digest worker enabled but CHANNEL_MEMORY_LLM_BASE_URL is unset; staying deterministic-only") + log.Printf("channel-memory digest worker enabled but CHANNEL_MEMORY_LLM_BASE_URL or CHANNEL_MEMORY_LLM_MODEL is unset; staying deterministic-only") } else { worker := newDigestWorker(store, client, workerCfg) go worker.Run(context.Background()) @@ -878,7 +878,47 @@ func markIdentityBlocksDirtyTx(ctx context.Context, tx *sql.Tx, sourceKind, chan args = append(args, exceptContentHash) } query += `)` - _, err := tx.ExecContext(ctx, query, args...) + if _, err := tx.ExecContext(ctx, query, args...); err != nil { + return err + } + return restoreCurrentRawExcerptsForDirtyRollupsTx(ctx, tx, sourceKind, channelID, messageID) +} + +func restoreCurrentRawExcerptsForDirtyRollupsTx(ctx context.Context, tx *sql.Tx, sourceKind, channelID, messageID string) error { + _, err := tx.ExecContext(ctx, ` + UPDATE derived_blocks + SET stale = 0 + WHERE processor = 'deterministic' AND kind = 'raw_excerpt' AND id IN ( + SELECT raw.block_id + FROM derived_block_sources raw + JOIN source_messages current_source + ON current_source.source_kind = raw.source_kind + AND current_source.channel_id = raw.channel_id + AND current_source.message_id = raw.message_id + AND current_source.content_hash = raw.content_hash + AND current_source.is_current = 1 + AND current_source.deleted = 0 + AND current_source.forgotten_at = '' + WHERE raw.source_kind = ? AND raw.channel_id = ? + AND raw.message_id IN ( + SELECT covered.message_id + FROM derived_blocks rollup + JOIN derived_block_sources trigger_source + ON trigger_source.block_id = rollup.id + AND trigger_source.source_kind = ? + AND trigger_source.channel_id = ? + AND trigger_source.message_id = ? + JOIN derived_block_sources covered + ON covered.block_id = rollup.id + WHERE rollup.processor = ? + AND rollup.kind IN (?, ?) + AND rollup.dirty = 1 + ) + )`, + sourceKind, channelID, + sourceKind, channelID, messageID, + digestProcessorLLM, rollupKindTopic, rollupKindSequence, + ) return err } @@ -1081,6 +1121,10 @@ func (s *channelMemoryStore) queryDigestBlocks(ctx context.Context, channelID, c if limit <= 0 { return nil, nil } + queryLimit := limit * 4 + if queryLimit < limit { + queryLimit = limit + } rows, err := s.db.QueryContext(ctx, ` SELECT id, kind, event_type, text, source_channel, source_window_from, source_window_to, sparse, score, generated_at, stale, dirty, processor @@ -1088,7 +1132,7 @@ func (s *channelMemoryStore) queryDigestBlocks(ctx context.Context, channelID, c WHERE source_channel = ? AND source_window_to >= ? AND stale = 0 AND dirty = 0 ORDER BY source_window_from ASC, id ASC LIMIT ?`, - channelID, cutoff, limit, + channelID, cutoff, queryLimit, ) if err != nil { return nil, err @@ -1124,7 +1168,48 @@ func (s *channelMemoryStore) queryDigestBlocks(ctx context.Context, channelID, c return nil, err } } - return blocks, nil + return preferSparseRollups(blocks, limit), nil +} + +func preferSparseRollups(blocks []digestBlock, limit int) []digestBlock { + coveredByRollup := make(map[string]struct{}) + for _, block := range blocks { + if block.Processor != digestProcessorLLM { + continue + } + if block.Kind != rollupKindTopic && block.Kind != rollupKindSequence { + continue + } + for _, messageID := range block.SourceMessages { + coveredByRollup[block.SourceChannel+"\x00"+messageID] = struct{}{} + } + } + if len(coveredByRollup) == 0 { + if len(blocks) > limit { + return blocks[:limit] + } + return blocks + } + filtered := make([]digestBlock, 0, len(blocks)) + for _, block := range blocks { + if block.Processor == "deterministic" && block.Kind == "raw_excerpt" { + skip := false + for _, messageID := range block.SourceMessages { + if _, ok := coveredByRollup[block.SourceChannel+"\x00"+messageID]; ok { + skip = true + break + } + } + if skip { + continue + } + } + filtered = append(filtered, block) + if len(filtered) >= limit { + break + } + } + return filtered } func (s *channelMemoryStore) loadBlockSources(ctx context.Context, block *digestBlock) error { diff --git a/examples/channel-memory/worker.go b/examples/channel-memory/worker.go index b10b8fd..7649d4c 100644 --- a/examples/channel-memory/worker.go +++ b/examples/channel-memory/worker.go @@ -36,6 +36,9 @@ const ( defaultWorkerMinWindow = 3 defaultWorkerPerChannelCalls = 24 defaultWorkerPerPodCalls = 96 + defaultWorkerPerChannelCost = 0.50 + defaultWorkerPerPodCost = 2.00 + defaultWorkerCostPerCall = 0.002 defaultWorkerIntervalSeconds = 60 defaultWorkerLLMTimeoutSecond = 30 ) @@ -75,6 +78,9 @@ type workerConfig struct { MinWindow int PerChannelDailyCalls int PerPodDailyCalls int + PerChannelDailyCost float64 + PerPodDailyCost float64 + CostPerCall float64 Interval time.Duration } @@ -96,6 +102,9 @@ func workerConfigFromEnv() workerConfig { MinWindow: intEnv("CHANNEL_MEMORY_LLM_MIN_WINDOW", defaultWorkerMinWindow), PerChannelDailyCalls: intEnv("CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY", defaultWorkerPerChannelCalls), PerPodDailyCalls: intEnv("CHANNEL_MEMORY_LLM_PER_POD_DAILY", defaultWorkerPerPodCalls), + PerChannelDailyCost: floatEnv("CHANNEL_MEMORY_LLM_PER_CHANNEL_DAILY_USD", defaultWorkerPerChannelCost), + PerPodDailyCost: floatEnv("CHANNEL_MEMORY_LLM_PER_POD_DAILY_USD", defaultWorkerPerPodCost), + CostPerCall: floatEnv("CHANNEL_MEMORY_LLM_COST_PER_CALL_USD", defaultWorkerCostPerCall), Interval: time.Duration(intEnv("CHANNEL_MEMORY_LLM_INTERVAL_SECONDS", defaultWorkerIntervalSeconds)) * time.Second, } if cfg.Provider == "" { @@ -120,6 +129,9 @@ func newDigestWorker(store *channelMemoryStore, client llmDigestClient, cfg work if cfg.Interval <= 0 { cfg.Interval = defaultWorkerIntervalSeconds * time.Second } + if cfg.CostPerCall < 0 { + cfg.CostPerCall = 0 + } return &digestWorker{ store: store, client: client, @@ -171,19 +183,18 @@ func (w *digestWorker) processOnce(ctx context.Context) (int, error) { day := w.now().UTC().Format("2006-01-02") generated := 0 for _, channelID := range channels { - podCalls, err := w.store.dailyLLMCalls(ctx, day, "pod") + podOK, err := w.podBudgetAvailable(ctx, day) if err != nil { return generated, err } - if podCalls >= w.cfg.PerPodDailyCalls { - w.logf("channel-memory digest worker: per-pod daily cap %d reached", w.cfg.PerPodDailyCalls) + if !podOK { break } - channelCalls, err := w.store.dailyLLMCalls(ctx, day, "channel:"+channelID) + channelOK, err := w.channelBudgetAvailable(ctx, day, channelID) if err != nil { return generated, err } - if channelCalls >= w.cfg.PerChannelDailyCalls { + if !channelOK { continue } @@ -192,18 +203,18 @@ func (w *digestWorker) processOnce(ctx context.Context) (int, error) { return generated, err } for _, window := range windows { - podCalls, err = w.store.dailyLLMCalls(ctx, day, "pod") + podOK, err = w.podBudgetAvailable(ctx, day) if err != nil { return generated, err } - if podCalls >= w.cfg.PerPodDailyCalls { + if !podOK { break } - channelCalls, err = w.store.dailyLLMCalls(ctx, day, "channel:"+channelID) + channelOK, err = w.channelBudgetAvailable(ctx, day, channelID) if err != nil { return generated, err } - if channelCalls >= w.cfg.PerChannelDailyCalls { + if !channelOK { break } @@ -218,12 +229,22 @@ func (w *digestWorker) processOnce(ctx context.Context) (int, error) { continue } + if err := w.store.recordLLMUsage(ctx, day, channelID, 0); err != nil { + return generated, err + } result, err := w.client.Summarize(ctx, promptForWindow(channelID, window)) if err != nil { w.logf("channel-memory digest worker: summarize channel %s: %v", channelID, err) _ = w.store.recordQueueFailure(ctx, sourceKind, channelID, window, err.Error()) continue } + costUSD := w.costForResult(result) + result.CostUSD = costUSD + if costUSD != 0 { + if err := w.store.recordLLMCost(ctx, day, channelID, costUSD); err != nil { + return generated, err + } + } if err := validateRollup(result, window); err != nil { w.logf("channel-memory digest worker: rejected rollup for channel %s: %v", channelID, err) _ = w.store.recordQueueFailure(ctx, sourceKind, channelID, window, err.Error()) @@ -238,15 +259,69 @@ func (w *digestWorker) processOnce(ctx context.Context) (int, error) { if err := w.store.writeSparseRollup(ctx, key, sourceKind, channelID, window, result, meta, w.now()); err != nil { return generated, err } - if err := w.store.recordLLMUsage(ctx, day, channelID, result.CostUSD); err != nil { - return generated, err - } generated++ } } return generated, nil } +func (w *digestWorker) podBudgetAvailable(ctx context.Context, day string) (bool, error) { + calls, err := w.store.dailyLLMCalls(ctx, day, "pod") + if err != nil { + return false, err + } + if calls >= w.cfg.PerPodDailyCalls { + w.logf("channel-memory digest worker: per-pod daily call cap %d reached", w.cfg.PerPodDailyCalls) + return false, nil + } + cost, err := w.store.dailyLLMCost(ctx, day, "pod") + if err != nil { + return false, err + } + if w.costWouldExceed(cost, w.cfg.PerPodDailyCost) { + w.logf("channel-memory digest worker: per-pod daily cost cap %.4f reached", w.cfg.PerPodDailyCost) + return false, nil + } + return true, nil +} + +func (w *digestWorker) channelBudgetAvailable(ctx context.Context, day, channelID string) (bool, error) { + scope := "channel:" + channelID + calls, err := w.store.dailyLLMCalls(ctx, day, scope) + if err != nil { + return false, err + } + if calls >= w.cfg.PerChannelDailyCalls { + return false, nil + } + cost, err := w.store.dailyLLMCost(ctx, day, scope) + if err != nil { + return false, err + } + if w.costWouldExceed(cost, w.cfg.PerChannelDailyCost) { + return false, nil + } + return true, nil +} + +func (w *digestWorker) costWouldExceed(current, cap float64) bool { + if cap <= 0 { + return false + } + projected := current + if w.cfg.CostPerCall > 0 { + projected += w.cfg.CostPerCall + } + return projected > cap +} + +func (w *digestWorker) costForResult(result llmDigestResult) float64 { + if result.CostUSD > 0 { + return result.CostUSD + } + return w.cfg.CostPerCall +} + type rollupWindow struct { Sources []storedSourceMessage } @@ -294,10 +369,23 @@ func validateRollup(result llmDigestResult, window rollupWindow) error { return errors.New("rollup is provenance-free (no source_messages)") } allowed := window.idSet() + seen := make(map[string]struct{}, len(cited)) for _, id := range cited { if _, ok := allowed[id]; !ok { return fmt.Errorf("rollup cites source %q outside its window", id) } + if _, ok := seen[id]; ok { + return fmt.Errorf("rollup cites source %q more than once", id) + } + seen[id] = struct{}{} + } + if len(seen) != len(allowed) { + for id := range allowed { + if _, ok := seen[id]; !ok { + return fmt.Errorf("rollup omits source %q from its window", id) + } + } + return fmt.Errorf("rollup cites %d sources, expected %d", len(seen), len(allowed)) } return nil } @@ -387,8 +475,9 @@ func (s *channelMemoryStore) freshBlockExists(ctx context.Context, blockKey stri } // writeSparseRollup persists the LLM rollup block with full source provenance -// and stales the per-message raw_excerpt blocks it subsumes, so /digest serves -// the compact rollup instead of the verbose lines. +// for /digest to prefer over the verbose per-message raw_excerpt blocks while +// it remains fresh. The deterministic raw blocks are left intact so they become +// the immediate fallback if the rollup is later dirtied or staled. func (s *channelMemoryStore) writeSparseRollup(ctx context.Context, blockKey, sourceKind, channelID string, window rollupWindow, result llmDigestResult, meta blockMetadata, now time.Time) error { if len(window.Sources) == 0 { return errors.New("empty rollup window") @@ -446,19 +535,6 @@ func (s *channelMemoryStore) writeSparseRollup(ctx context.Context, blockKey, so ); err != nil { return err } - // Stale the verbose deterministic raw_excerpt block this rollup - // subsumes. Hard events / tombstones / telemetry blocks are untouched. - if _, err := tx.ExecContext(ctx, ` - UPDATE derived_blocks - SET stale = 1 - WHERE processor = 'deterministic' AND kind = 'raw_excerpt' AND id IN ( - SELECT block_id FROM derived_block_sources - WHERE source_kind = ? AND channel_id = ? AND message_id = ? AND content_hash = ? - ) AND id <> ?`, - src.SourceKind, src.ChannelID, src.MessageID, src.ContentHash, blockID, - ); err != nil { - return err - } } return tx.Commit() } @@ -478,6 +554,21 @@ func (s *channelMemoryStore) dailyLLMCalls(ctx context.Context, day, scope strin return calls, nil } +func (s *channelMemoryStore) dailyLLMCost(ctx context.Context, day, scope string) (float64, error) { + var cost float64 + err := s.db.QueryRowContext(ctx, + `SELECT cost_usd FROM llm_usage WHERE day = ? AND scope = ?`, + day, scope, + ).Scan(&cost) + if errors.Is(err, sql.ErrNoRows) { + return 0, nil + } + if err != nil { + return 0, err + } + return cost, nil +} + func (s *channelMemoryStore) recordLLMUsage(ctx context.Context, day, channelID string, costUSD float64) error { for _, scope := range []string{"pod", "channel:" + channelID} { if _, err := s.db.ExecContext(ctx, ` @@ -494,6 +585,21 @@ func (s *channelMemoryStore) recordLLMUsage(ctx context.Context, day, channelID return nil } +func (s *channelMemoryStore) recordLLMCost(ctx context.Context, day, channelID string, costUSD float64) error { + for _, scope := range []string{"pod", "channel:" + channelID} { + if _, err := s.db.ExecContext(ctx, ` + INSERT INTO llm_usage(day, scope, calls, cost_usd) + VALUES (?, ?, 0, ?) + ON CONFLICT(day, scope) DO UPDATE SET + cost_usd = cost_usd + excluded.cost_usd`, + day, scope, costUSD, + ); err != nil { + return err + } + } + return nil +} + func (s *channelMemoryStore) recordQueueFailure(ctx context.Context, sourceKind, channelID string, window rollupWindow, message string) error { now := s.now().UTC().Format(time.RFC3339) for _, src := range window.Sources { @@ -529,7 +635,7 @@ type httpLLMClient struct { func httpLLMClientFromEnv(cfg workerConfig) llmDigestClient { baseURL := strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_LLM_BASE_URL")) - if baseURL == "" { + if baseURL == "" || strings.TrimSpace(cfg.Model) == "" { return nil } return &httpLLMClient{ @@ -555,7 +661,7 @@ func (c *httpLLMClient) Summarize(ctx context.Context, prompt llmDigestPrompt) ( {"role": "user", "content": "Channel " + prompt.Channel + " transcript:\n" + transcript.String()}, }, "response_format": map[string]string{"type": "json_object"}, - "temperature": "0", + "temperature": 0, } body, err := json.Marshal(payload) if err != nil { @@ -624,3 +730,15 @@ func intEnv(name string, fallback int) int { } return value } + +func floatEnv(name string, fallback float64) float64 { + raw := strings.TrimSpace(os.Getenv(name)) + if raw == "" { + return fallback + } + value, err := strconv.ParseFloat(raw, 64) + if err != nil || value < 0 { + return fallback + } + return value +} diff --git a/examples/channel-memory/worker_test.go b/examples/channel-memory/worker_test.go index 14835b4..3ff770d 100644 --- a/examples/channel-memory/worker_test.go +++ b/examples/channel-memory/worker_test.go @@ -207,6 +207,9 @@ func TestDigestWorkerRejectsMalformedAndProvenanceFreeOutput(t *testing.T) { {"provenance free", func(p llmDigestPrompt) (llmDigestResult, error) { return llmDigestResult{Kind: rollupKindTopic, Text: "summary", SourceMessages: nil}, nil }}, + {"partial provenance", func(p llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{Kind: rollupKindTopic, Text: "summary", SourceMessages: []string{p.Messages[0].MessageID}}, nil + }}, {"hallucinated source", func(p llmDigestPrompt) (llmDigestResult, error) { return llmDigestResult{Kind: rollupKindTopic, Text: "summary", SourceMessages: []string{"999999"}}, nil }}, @@ -338,6 +341,49 @@ func TestDigestWorkerEnforcesDailyCostCaps(t *testing.T) { } } +// TestDigestWorkerEnforcesDailyUSDCaps proves the USD cap is separate from the +// call cap and can stop a second window even when more calls are allowed. +func TestDigestWorkerEnforcesDailyUSDCaps(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + for i := 0; i < 6; i++ { + id := fmt.Sprintf("45%d", i) + ingestRaw(t, store, "chan-1", id, "idle chatter "+id, base.Add(time.Duration(i)*time.Minute), "sha256:r"+id) + } + + client := &fakeLLMClient{respond: func(p llmDigestPrompt) (llmDigestResult, error) { + result, err := citeAllTopicRollup(p) + result.CostUSD = 0.30 + return result, err + }} + worker := testWorker(store, client, func(c *workerConfig) { + c.PerPodDailyCalls = 100 + c.PerChannelDailyCalls = 100 + c.PerPodDailyCost = 0.50 + c.PerChannelDailyCost = 0.50 + c.CostPerCall = 0.30 + }) + + generated, err := worker.processOnce(context.Background()) + if err != nil { + t.Fatalf("processOnce: %v", err) + } + if generated != 1 { + t.Fatalf("expected USD cap to allow exactly 1 block, got %d", generated) + } + if client.callCount() != 1 { + t.Fatalf("expected USD cap to allow exactly 1 LLM call, got %d", client.callCount()) + } + cost, err := store.dailyLLMCost(context.Background(), worker.now().UTC().Format("2006-01-02"), "pod") + if err != nil { + t.Fatalf("daily cost: %v", err) + } + if cost < 0.2999 || cost > 0.3001 { + t.Fatalf("expected estimated cost 0.30, got %.4f", cost) + } +} + // TestDigestWorkerDeterministicFallbackWhenDisabledOrFailing proves the worker // is inert when disabled and harmless when the model errors. func TestDigestWorkerDeterministicFallbackWhenDisabledOrFailing(t *testing.T) { @@ -396,6 +442,47 @@ func TestDigestWorkerDeterministicFallbackWhenDisabledOrFailing(t *testing.T) { }) } +// TestDigestWorkerCountsRejectedCallsAgainstDailyCaps proves invalid or failing +// model calls still consume the call budget. Otherwise a bad provider response +// can be retried every worker interval without tripping the cap. +func TestDigestWorkerCountsRejectedCallsAgainstDailyCaps(t *testing.T) { + store := newTestStore(t) + defer store.Close() + base := time.Date(2026, 5, 21, 18, 0, 0, 0, time.UTC) + ingestRaw(t, store, "chan-1", "551", "idle one", base, "sha256:r551") + ingestRaw(t, store, "chan-1", "552", "idle two", base.Add(time.Minute), "sha256:r552") + ingestRaw(t, store, "chan-1", "553", "idle three", base.Add(2*time.Minute), "sha256:r553") + + client := &fakeLLMClient{respond: func(p llmDigestPrompt) (llmDigestResult, error) { + return llmDigestResult{Kind: rollupKindTopic, Text: "summary", SourceMessages: nil}, nil + }} + worker := testWorker(store, client, func(c *workerConfig) { + c.PerPodDailyCalls = 1 + c.PerChannelDailyCalls = 1 + }) + + if generated, err := worker.processOnce(context.Background()); err != nil || generated != 0 { + t.Fatalf("first processOnce generated=%d err=%v", generated, err) + } + if client.callCount() != 1 { + t.Fatalf("expected first rejected output to call model once, got %d", client.callCount()) + } + resp := digestFor(t, store, "chan-1") + if resp.Cost.LLMCallsToday != 1 { + t.Fatalf("expected rejected call to count against daily cap, got %d", resp.Cost.LLMCallsToday) + } + if !resp.Cost.DeterministicOnly { + t.Fatal("rejected output must leave digest deterministic-only") + } + + if generated, err := worker.processOnce(context.Background()); err != nil || generated != 0 { + t.Fatalf("second processOnce generated=%d err=%v", generated, err) + } + if client.callCount() != 1 { + t.Fatalf("expected daily cap to prevent retry after rejection, got %d calls", client.callCount()) + } +} + // TestDigestWorkerEditAndForgetMarkRollupStaleOrDirty proves provenance-driven // invalidation: editing or forgetting a covered source dirties the rollup so it // stops serving until regenerated. Uses synthetic source events because @@ -422,10 +509,17 @@ func TestDigestWorkerEditAndForgetMarkRollupStaleOrDirty(t *testing.T) { t.Fatalf("expected rollup dirtied after edit, found=%v dirty=%v", found, dirty) } // Dirty blocks are not served. + rawCount := 0 for _, b := range digestFor(t, store, "chan-1").Blocks { if b.Processor == digestProcessorLLM { t.Fatal("dirty rollup must not be served") } + if b.Kind == "raw_excerpt" { + rawCount++ + } + } + if rawCount != 3 { + t.Fatalf("expected deterministic raw_excerpt fallback after edit, got %d", rawCount) } }) @@ -447,5 +541,20 @@ func TestDigestWorkerEditAndForgetMarkRollupStaleOrDirty(t *testing.T) { if !found || !dirty { t.Fatalf("expected rollup dirtied after forget, found=%v dirty=%v", found, dirty) } + got := make(map[string]bool) + for _, b := range digestFor(t, store, "chan-1").Blocks { + if b.Processor == digestProcessorLLM { + t.Fatal("dirty rollup must not be served after forget") + } + if b.Kind == "raw_excerpt" && len(b.SourceMessages) == 1 { + got[b.SourceMessages[0]] = true + } + } + if got["702"] { + t.Fatalf("forgotten source 702 must not return in fallback: %v", got) + } + if !got["701"] || !got["703"] || len(got) != 2 { + t.Fatalf("expected current raw fallback for 701 and 703 after forget, got %v", got) + } }) }