diff --git a/ARCHITECTURE.md b/docs/ARCHITECTURE.md similarity index 100% rename from ARCHITECTURE.md rename to docs/ARCHITECTURE.md diff --git a/docs/CHANGES-0.15.0-phase1.md b/docs/CHANGES-0.15.0-phase1.md new file mode 100644 index 0000000..492ccaf --- /dev/null +++ b/docs/CHANGES-0.15.0-phase1.md @@ -0,0 +1,93 @@ +# v0.15.0 Phase 1 — Changeset Notes + +## Files included in this zip (new or fully replaced) + +``` +server/treepath/path.go NEW — PathMessage, GetActivePath, etc. +server/treepath/summary.go NEW — IsSummaryMessage, FindSummaryBoundary +server/treepath/cursor.go NEW — UpdateCursor, NextSiblingIndex +server/treepath/siblings.go NEW — GetSiblingCount, GetSiblings, FindLeafFromMessage +server/compaction/compaction.go NEW — Service, Compact, CheckRateLimit +server/handlers/tree.go REPLACED — thin wrappers delegating to treepath +server/handlers/summarize.go REPLACED — thin HTTP wrapper using compaction.Service +src/js/settings-handlers.js REPLACED — bug fixes for display name + role config +``` + +## Surgical edits needed in existing files + +### 1. `server/handlers/completion.go` + +**Delete** the `isSummaryMessage` function (lines ~832–841). It is now +defined in `treepath/summary.go` and aliased in `handlers/tree.go`. +Without this deletion, the build fails with a duplicate function error. + +```go +// DELETE this entire function from completion.go: + +// isSummaryMessage checks if a PathMessage has summary metadata. +func isSummaryMessage(m *PathMessage) bool { + if m.Metadata == nil { + return false + } + var meta map[string]interface{} + if err := json.Unmarshal(*m.Metadata, &meta); err != nil { + return false + } + return meta["type"] == "summary" +} +``` + +No other changes needed in completion.go — the `isSummaryMessage` calls +throughout the file now resolve via the alias in `handlers/tree.go`. + +### 2. `server/main.go` + +**Change** the summarize handler wiring (~line 252). Old constructor +takes `(stores, roleResolver)`, new one takes `(*compaction.Service)`. + +```go +// ── BEFORE ── +// Summarize & Continue +summarize := handlers.NewSummarizeHandler(stores, roleResolver) +protected.POST("/channels/:id/summarize", summarize.Summarize) + +// ── AFTER ── +// Compaction service (shared by manual summarize handler + future auto-scanner) +compactionSvc := compaction.NewService(stores, roleResolver) + +// Summarize & Continue (manual compaction via HTTP) +summarize := handlers.NewSummarizeHandler(compactionSvc) +protected.POST("/channels/:id/summarize", summarize.Summarize) +``` + +**Add import** at top of main.go: +```go +"git.gobha.me/xcaliber/chat-switchboard/compaction" +``` + +## Bug fixes included + +### Display Name not saved or used + +`handleSaveSettings()` now also saves `display_name` and `email` via +`API.updateProfile()`. Updates `API.user` and calls `UI.updateUser()` so +the sidebar reflects the change immediately. + +### User utility model role not displayed after save + +`_initUserRolePrimitive()` → `fetchModels` callback now calls the +global `fetchModels()` (which refreshes `App.models` from the server) +before returning the model list. This ensures recently added personal +provider models appear in the role config dropdown after tab switch or +save-refresh. + +## Verification + +After applying all changes: + +1. `cd server && go build .` — should compile cleanly +2. Existing integration tests should pass (no behavior change) +3. Manual test: Settings → Profile → change display name → Save → verify + sidebar updates and value persists on reload +4. Manual test: Settings → Model Roles → pick personal provider + model → + Save → verify dropdown shows saved values after refresh diff --git a/docs/CHANGES-0.15.0-phase2.md b/docs/CHANGES-0.15.0-phase2.md new file mode 100644 index 0000000..c44f5b7 --- /dev/null +++ b/docs/CHANGES-0.15.0-phase2.md @@ -0,0 +1,186 @@ +# v0.15.0 Phase 2 — Changeset Notes + +## Files included in this zip (new) + +``` +server/compaction/estimator.go NEW — EstimateTokens, EstimatePath, EstimatePathFromSummary +server/compaction/estimator_test.go NEW — unit tests (no DB required) +server/compaction/scanner.go NEW — Scanner, scan loop, candidate query, shouldCompact +``` + +## Surgical edits needed in existing files + +### 1. `server/main.go` + +Phase 1 already introduced `compactionSvc`. Now add the scanner below it +(after the `compactionSvc` line, before route registration): + +```go +import "git.gobha.me/xcaliber/chat-switchboard/compaction" // already present from Phase 1 + +// ── Auto-Compaction Scanner ─────────── +compactionScanner := compaction.NewScanner(compactionSvc, stores, compaction.ScannerConfig{ + Interval: 5 * time.Minute, + Concurrency: 2, +}) +compactionScanner.Start() +defer compactionScanner.Stop() // drain in-flight compactions on shutdown +``` + +Place `defer compactionScanner.Stop()` near the existing +`defer kbIngester.Wait()` line. + +**Note:** The scanner re-reads all config from `global_settings` each +tick. The `ScannerConfig` struct only sets the ticker interval and +semaphore size at startup. Everything else (enabled, threshold, cooldown) +is dynamic. + +--- + +### 2. `src/index.html` + +Add an auto-compaction section to the admin Settings tab. Insert **before** +the Encryption section (before `
` with +`🔐 Encryption`): + +```html +
+

Auto-Compaction

+ +

When enabled, long conversations are automatically summarized in the background using the utility model. Ships off by default — requires a utility model role to be configured.

+ +
+``` + +--- + +### 3. `src/js/ui-admin.js` — `loadAdminSettings()` + +Add these lines **after** the Web Search config loading block +(after `document.getElementById('searxngConfigFields').style.display = ...`): + +```javascript + // Auto-Compaction (global_settings) + const compactionCfg = getSetting('auto_compaction', {}) || {}; + const compactionEnabled = document.getElementById('adminCompactionEnabled'); + if (compactionEnabled) { + compactionEnabled.checked = !!compactionCfg.enabled; + document.getElementById('compactionConfigFields').style.display = + compactionCfg.enabled ? '' : 'none'; + } + const compThreshold = document.getElementById('adminCompactionThreshold'); + if (compThreshold) compThreshold.value = String(compactionCfg.threshold || 70); + const compCooldown = document.getElementById('adminCompactionCooldown'); + if (compCooldown) compCooldown.value = String(compactionCfg.cooldown || 30); +``` + +--- + +### 4. `src/js/settings-handlers.js` — `handleSaveAdminSettings()` + +Add these lines **after** the `search_config` save block (before +`UI.toast('Settings saved', 'success')`): + +```javascript + // Auto-Compaction config → global_settings + const compactionEnabled = document.getElementById('adminCompactionEnabled')?.checked || false; + const compactionThreshold = parseInt(document.getElementById('adminCompactionThreshold')?.value) || 70; + const compactionCooldown = parseInt(document.getElementById('adminCompactionCooldown')?.value) || 30; + await API.adminUpdateSetting('auto_compaction', { value: { + enabled: compactionEnabled, + threshold: compactionThreshold, + cooldown: compactionCooldown, + }}); + + // Persist individual keys for scanner (reads these independently) + await API.adminUpdateSetting('auto_compaction_enabled', { value: compactionEnabled }); + await API.adminUpdateSetting('auto_compaction_threshold', { value: compactionThreshold / 100 }); + await API.adminUpdateSetting('auto_compaction_cooldown_minutes', { value: compactionCooldown }); +``` + +--- + +### 5. `src/js/settings-handlers.js` — `_initSettingsListeners()` + +Add a toggle listener for the compaction checkbox, alongside the existing +banner toggle. Add **after** the `adminBannerEnabled` change listener: + +```javascript + document.getElementById('adminCompactionEnabled')?.addEventListener('change', (e) => { + document.getElementById('compactionConfigFields').style.display = e.target.checked ? '' : 'none'; + }); +``` + +--- + +### 6. `src/js/ui-core.js` — `_summaryHTML()` + +Show "(auto)" label on auto-triggered summaries. The `trigger` field is +already in message metadata from Phase 1. + +**Replace** the `_summaryHTML` method: + +```javascript + _summaryHTML(msg) { + const meta = msg.metadata || {}; + const count = meta.summarized_count || '?'; + const model = meta.utility_model || 'utility model'; + const trigger = meta.trigger === 'auto' ? ' · auto' : ''; + return ` +
+
+ 📝 Conversation summary (${count} messages, by ${esc(model)}${trigger}) +
+
${formatMessage(msg.content)}
+
`; + }, +``` + +--- + +## How the scanner reads settings + +The scanner calls `GlobalConfig.Get()` **every tick** for these keys: + +| Key | Read by | How stored | +|-----|---------|------------| +| `auto_compaction_enabled` | `isEnabled()` | `{value: true/false}` | +| `auto_compaction_threshold` | `getThreshold()` | `{value: 0.70}` (float 0-1) | +| `auto_compaction_cooldown_minutes` | `getCooldownDuration()` | `{value: 30}` | + +The admin UI stores a combined `auto_compaction` key (for display) AND +the individual keys (for scanner consumption). This avoids the scanner +needing to parse a composite object. + +Interval and concurrency are set at startup via `ScannerConfig` and +require a restart to change. This is intentional — they affect the +goroutine pool size and ticker, which shouldn't be hot-swapped. + +## Verification + +1. `cd server && go test ./compaction/` — estimator tests pass (no DB) +2. `cd server && go build .` — compiles cleanly +3. Manual test: Admin → Settings → enable auto-compaction → Save → + check server logs for `🔍 compaction scanner started` +4. Manual test: with utility role configured and a long conversation, + verify auto-compaction fires after the threshold is exceeded +5. Verify cooldown: same channel should not re-compact within 30 minutes +6. Verify kill switch: disable auto-compaction → scanner stops processing + on next tick diff --git a/docs/CHANGES-0.15.0-phase4.md b/docs/CHANGES-0.15.0-phase4.md new file mode 100644 index 0000000..deb3147 --- /dev/null +++ b/docs/CHANGES-0.15.0-phase4.md @@ -0,0 +1,106 @@ +# v0.15.0 Phase 4 — Guard Rail + Integration Tests + +## Files + +``` +server/compaction/compaction.go UPDATED — context budget guard rail +server/compaction/compaction_test.go NEW — scanner integration tests (DB required) +server/compaction/testmain_test.go NEW — TestMain for compaction package +server/database/seed_helpers.go NEW — SeedTestMessage, SeedTestMessages, SeedTestCursor +server/handlers/live_compaction_test.go NEW — e2e tests with Venice/Qwen3-4B +``` + +## Changes + +### 1. Context Budget Guard Rail (`compaction.go`) + +Added between prompt construction and LLM call. Prevents sending +truncated input to small utility models. + +**How it works:** +1. Estimates total prompt tokens (system + conversation content) +2. Resolves the utility model's `max_context` from catalog +3. Reserves 20% for output (the summary itself) +4. If `estimated_prompt > max_context × 0.80`, returns `ErrContextBudget` + +**Resolution chain** for utility model context: +- `Resolver.GetConfig()` → primary binding → `Catalog.GetByModelID()` +- Fallback: `Catalog.GetByModelIDAny()` (different provider) +- Hard fallback: `DefaultBudget` (128K) + +The scanner's `doCompact()` already handles errors gracefully — a budget +error just logs a warning and retries next cycle (by which time the +conversation may have been manually compacted or the admin may have +upgraded the utility model). + +**New export:** `ErrContextBudget` — callers can use `errors.Is()` to +distinguish budget failures from LLM errors. + +### 2. Test Seed Helpers (`database/seed_helpers.go`) + +New file in the `database` package alongside `testhelper.go`: + +- `SeedTestMessage(t, channelID, parentID, role, content) → msgID` +- `SeedTestMessages(t, channelID, count, contentSize) → []msgIDs` + Creates a linear alternating user/assistant chain +- `SeedTestCursor(t, channelID, userID, leafID)` + Sets the active leaf for tree path resolution + +### 3. Scanner Integration Tests (`compaction/compaction_test.go`) + +All DB-dependent tests use `database.RequireTestDB(t)` and skip +gracefully when no DB is configured. + +| Test | What it verifies | +|------|-----------------| +| `FindCandidates_ReturnsQualifying` | Channel with ≥10 msgs, ≥20K chars, proper age range appears | +| `FindCandidates_ExcludesTooFewMessages` | <10 messages → excluded | +| `FindCandidates_ExcludesTooRecent` | Updated just now → excluded (activity gap) | +| `FindCandidates_ExcludesArchived` | `is_archived=true` → excluded | +| `Cooldown` | Timestamps recorded and checked correctly | +| `InFlightDedup` | sync.Map prevents concurrent compaction of same channel | +| `ShouldCompact_ChannelOptOut` | `settings.auto_compaction=false` → rejected | +| `IsEnabled` | Reads `auto_compaction_enabled` from global_settings | +| `GetThreshold_Default` | Returns 0.70 when no override | +| `GetThreshold_GlobalOverride` | Reads from global_settings | +| `GetThreshold_ChannelOverride` | Channel setting takes precedence | +| `GetCooldownDuration_Default` | Returns 30min default | +| `GetCooldownDuration_Override` | Reads from global_settings | +| `GuardRailMath` | Validates token math for 32K vs 128K models | + +### 4. Live E2E Tests (`handlers/live_compaction_test.go`) + +Requires: `TEST_DATABASE_URL` + `VENICE_API_KEY` + +| Test | What it verifies | +|------|-----------------| +| `CompactionFullPipeline` | Seeds 10 realistic messages → Compact() → verifies summary message in DB, metadata fields, cursor update, usage log entry | +| `CompactionContextBudgetGuardRail` | Seeds 160K chars of content → Compact() returns ErrContextBudget with 32K model | + +## Running Tests + +```bash +# Unit only (no DB) +cd server && go test ./compaction/ -run TestEstimate -v + +# Integration (requires DB) +cd server && go test ./compaction/ -v + +# Live e2e (requires DB + Venice key) +cd server && VENICE_API_KEY=... go test ./handlers/ -run TestLive_Compaction -v +``` + +## Surgical Edit + +### `database/testhelper.go` + +Add this import if not already present (the new `seed_helpers.go` file +is a separate file in the same package, so no edit needed to +`testhelper.go` itself): + +```go +// No changes needed — seed_helpers.go is a new file in package database +``` + +The `strings` import in `seed_helpers.go` is used by `SeedTestMessages` +for `strings.Repeat`. diff --git a/DESIGN-0.12.0.md b/docs/DESIGN-0.12.0.md similarity index 100% rename from DESIGN-0.12.0.md rename to docs/DESIGN-0.12.0.md diff --git a/DESIGN-0.13.1.md b/docs/DESIGN-0.13.1.md similarity index 100% rename from DESIGN-0.13.1.md rename to docs/DESIGN-0.13.1.md diff --git a/DESIGN-0.14.0.md b/docs/DESIGN-0.14.0.md similarity index 100% rename from DESIGN-0.14.0.md rename to docs/DESIGN-0.14.0.md diff --git a/docs/DESIGN-0.15.0.md b/docs/DESIGN-0.15.0.md new file mode 100644 index 0000000..b6c8e7a --- /dev/null +++ b/docs/DESIGN-0.15.0.md @@ -0,0 +1,584 @@ +# DESIGN-0.15.0 — Compaction + +## Overview + +Automatic conversation compaction. A background service monitors channels +for context pressure and triggers summarization via the utility model role, +replacing the manual "Summarize & Continue" button as the primary +compaction path. Manual summarization remains available as an explicit +user action. + +Depends on: utility model role (v0.10.0), summarize handler (v0.10.2). + +**Design principle: don't reinvent the wheel.** The existing +`SummarizeHandler` already has the full pipeline — path loading, summary +boundary detection, prompt construction, role resolution, tree insertion, +cursor update, usage logging. This design extracts that core logic into +a shared `compaction` package and adds a background scanner on top. + +--- + +## 1. Extract: `compaction` Package + +Move the reusable summarization pipeline out of `handlers/summarize.go` +into `server/compaction/compaction.go`. The HTTP handler becomes a thin +wrapper. + +### Core type + +```go +package compaction + +// Service provides conversation compaction (summarization). +// Used by both the HTTP handler (manual) and the background scanner (auto). +type Service struct { + stores store.Stores + resolver *roles.Resolver +} + +func NewService(stores store.Stores, resolver *roles.Resolver) *Service { + return &Service{stores: stores, resolver: resolver} +} +``` + +### Extracted method: `Compact` + +The current `SummarizeHandler.Summarize` body becomes `Service.Compact`: + +```go +type CompactRequest struct { + ChannelID string + UserID string // channel owner + TeamID *string // for role resolution + Trigger string // "manual" | "auto" +} + +type CompactResult struct { + SummaryID string + SummarizedCount int + Model string + UsedFallback bool + Content string + InputTokens int + OutputTokens int +} + +func (s *Service) Compact(ctx context.Context, req CompactRequest) (*CompactResult, error) +``` + +The method performs the same steps as today's handler: + +1. Load active path via `getActivePath` (already exported within `handlers`) +2. Find summary boundary, collect messages to summarize +3. Guard: minimum 4 messages since last summary +4. Build summarization prompt +5. Call `resolver.Complete(ctx, RoleUtility, ...)` +6. Insert summary tree node with metadata +7. Update cursor +8. Log usage + +The only new metadata field is `"trigger"` (`"manual"` or `"auto"`) so +the frontend can distinguish how the summary was created. + +### Refactored handler + +`handlers/summarize.go` shrinks to: + +```go +func (h *SummarizeHandler) Summarize(c *gin.Context) { + // ownership check, rate limit check (unchanged) + // ... + result, err := h.compaction.Compact(c.Request.Context(), compaction.CompactRequest{ + ChannelID: channelID, + UserID: userID, + TeamID: teamID, + Trigger: "manual", + }) + // return JSON response (unchanged) +} +``` + +### Tree helpers + +`getActivePath`, `isSummaryMessage`, `nextSiblingIndex`, `updateCursor` +are currently unexported in `handlers/tree.go`. Two options: + +**Option A** — Export them from `handlers` and import in `compaction`. +Clean but creates a dependency from `compaction` → `handlers`. + +**Option B** — Move tree helpers into a `treepath` (or similar) package +imported by both `handlers` and `compaction`. + +Recommend **Option B** to keep the dependency graph clean: +`handlers` → `compaction` → `treepath` ← `handlers`. The `treepath` +package is pure data + SQL queries with no handler logic. + +``` +server/ + treepath/ ← NEW: extracted from handlers/tree.go + path.go (getActivePath, getPathToLeaf, getActiveLeaf) + summary.go (isSummaryMessage, summary metadata helpers) + siblings.go (getSiblingCount, getSiblings, findLeafFromMessage) + cursor.go (updateCursor, nextSiblingIndex) + compaction/ ← NEW + compaction.go (Service, Compact) + scanner.go (Scanner, background loop) + estimator.go (token estimation) + handlers/ + tree.go → thin wrappers or deleted, imports treepath + summarize.go → thin HTTP wrapper, delegates to compaction.Service + completion.go → imports treepath for path building +``` + +--- + +## 2. Server-Side Token Estimation + +The background scanner needs to evaluate context pressure without making +an LLM call. The frontend already does this with a `chars / 4` heuristic +(`tokens.js`). Replicate server-side: + +```go +// compaction/estimator.go + +// EstimateTokens returns a rough token count using the ~4 chars/token +// heuristic. Matches the frontend Tokens.estimate() for consistency. +func EstimateTokens(text string) int { + return (len(text) + 3) / 4 +} + +// EstimatePath returns total estimated tokens for a message path, +// including system prompt overhead. +func EstimatePath(path []treepath.PathMessage, systemPromptLen int) int { + total := 0 + if systemPromptLen > 0 { + total += EstimateTokens(string(make([]byte, systemPromptLen))) + 4 + } + for _, m := range path { + total += EstimateTokens(m.Content) + 4 // +4 per-message overhead + } + return total +} +``` + +Not exact, but it's the same bar the user already sees in the UI. If we +need better accuracy later, swap in a tokenizer library without changing +the interface. + +### Context budget resolution + +The scanner needs `max_context` for whatever model the channel is using. +Resolution chain: + +1. Channel has a `model` field → look up `model_catalog` for capabilities +2. Channel has a `provider_config_id` → use that provider's catalog entry +3. Fallback: use the utility role's model context window as a proxy (it's + the model that would be doing the summarization anyway) +4. Hard fallback: 128K tokens (conservative default) + +```go +func (s *Scanner) getContextBudget(ctx context.Context, ch *models.Channel) int { + if ch.Model != "" { + if caps, err := s.stores.Catalog.GetCapabilities(ctx, ch.Model); err == nil { + if caps.MaxContext > 0 { + return caps.MaxContext + } + } + } + return DefaultContextBudget // 128000 +} +``` + +--- + +## 3. Background Scanner + +Follows the Ingester pattern from `knowledge/ingest.go`: goroutine pool +with semaphore, WaitGroup for graceful shutdown. + +### Scanner type + +```go +// compaction/scanner.go + +type Scanner struct { + service *Service + stores store.Stores + sem chan struct{} + wg sync.WaitGroup + stopCh chan struct{} + ticker *time.Ticker +} + +type ScannerConfig struct { + Enabled bool // global kill switch + Interval time.Duration // scan frequency (default: 5 minutes) + Concurrency int // max parallel compactions (default: 2) +} + +func NewScanner(svc *Service, stores store.Stores, cfg ScannerConfig) *Scanner +func (sc *Scanner) Start() +func (sc *Scanner) Stop() // signals stop + drains in-flight +``` + +### Scan loop + +``` +every : + if !globalEnabled(): continue + + candidates = findCandidates() + for each candidate: + sem.acquire() + go compact(candidate) +``` + +### Candidate query + +A single SQL query finds channels that need compaction: + +```sql +SELECT c.id, c.user_id, c.model, c.settings, + COUNT(m.id) as msg_count, + SUM(LENGTH(m.content)) as total_chars +FROM channels c +JOIN messages m ON m.channel_id = c.id + AND m.deleted_at IS NULL +WHERE c.deleted_at IS NULL + AND c.type = 'direct' -- only user conversations (not groups yet) + AND c.is_archived = false + -- Skip channels with very recent activity (let the user finish typing) + AND c.updated_at < NOW() - INTERVAL '2 minutes' + -- Only channels active in the last 7 days (don't compact stale channels) + AND c.updated_at > NOW() - INTERVAL '7 days' +GROUP BY c.id +HAVING COUNT(m.id) >= 10 -- minimum message count + AND SUM(LENGTH(m.content)) > 20000 -- rough: 20K chars ≈ 5K tokens minimum +ORDER BY c.updated_at DESC +LIMIT 50 -- batch size cap +``` + +This is a coarse filter. Each candidate then gets a precise check: + +```go +func (sc *Scanner) shouldCompact(ctx context.Context, ch *models.Channel) bool { + // 1. Check channel-level opt-out + if ch.Settings != nil { + if v, ok := ch.Settings["auto_compaction"]; ok && v == false { + return false + } + } + + // 2. Load active path for the channel owner + path, err := treepath.GetActivePath(ch.ID, ch.UserID) + if err != nil || len(path) < 10 { + return false + } + + // 3. Find post-summary messages only + startIdx := 0 + for i, m := range path { + if treepath.IsSummaryMessage(&m) { + startIdx = i + 1 + } + } + postSummary := path[startIdx:] + if len(postSummary) < 8 { + return false // not enough new messages since last summary + } + + // 4. Estimate tokens and check threshold + tokens := EstimatePath(postSummary, 0) + budget := sc.getContextBudget(ctx, ch) + threshold := sc.getThreshold(ch) // default 0.70 + + return float64(tokens) / float64(budget) >= threshold +} +``` + +### Threshold + +Default: **0.70** (70% of context window). This is intentionally lower +than the frontend's 75% warning bar — auto-compaction should fire before +the user sees a warning, not after. + +Configurable per-channel via `settings.compaction_threshold` (float) and +globally via `global_settings` key `auto_compaction_threshold`. + +### Rate limiting + +Auto-compaction calls use the utility role and are org-funded. They share +the existing `utility_rate_limit` budget but get a separate counter +prefix so admins can see auto vs manual usage: + +```go +entry := &models.UsageEntry{ + // ... + Role: &role, // "utility" + Metadata: models.JSONMap{ + "trigger": "auto", // distinguishes from manual + }, +} +``` + +If the global utility rate limit is hit, auto-compaction silently skips +the channel and retries next scan cycle. It never blocks or errors. + +--- + +## 4. Per-Channel Configuration + +Uses the existing `Channel.Settings` JSONB column. No migration needed. + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `auto_compaction` | `bool` | `true` (from global) | Enable/disable for this channel | +| `compaction_threshold` | `float` | `0.70` (from global) | Context usage ratio to trigger | + +### Frontend: channel settings + +Add a "Compaction" section to the channel settings panel (the gear icon +in the chat header). Two controls: + +``` +┌─ Compaction ──────────────────────────────────┐ +│ Auto-compact [✓] │ +│ Threshold [70]% of context window │ +└───────────────────────────────────────────────┘ +``` + +The manual "Summarize & Continue" button remains in the context warning +bar and works regardless of auto-compaction settings. + +--- + +## 5. Admin Controls + +New `global_settings` keys: + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `auto_compaction_enabled` | `bool` | `false` | Global kill switch. Ships **off** — admins opt in. | +| `auto_compaction_threshold` | `float` | `0.70` | Default threshold for channels without override | +| `auto_compaction_interval_minutes` | `int` | `5` | Scanner tick interval | +| `auto_compaction_concurrency` | `int` | `2` | Max parallel auto-compactions | +| `auto_compaction_cooldown_minutes` | `int` | `30` | Min time between auto-compactions of the same channel | + +### Admin UI + +Add an "Auto-Compaction" card to the existing admin AI settings panel +(alongside the utility rate limit and model roles): + +``` +┌─ Auto-Compaction ─────────────────────────────┐ +│ Enabled [✓] │ +│ Threshold [70]% │ +│ Scan interval [5] minutes │ +│ Concurrency [2] parallel │ +│ Cooldown [30] minutes per channel │ +└───────────────────────────────────────────────┘ +``` + +--- + +## 6. Cooldown + Dedup + +Prevent rapid re-compaction of the same channel: + +1. **Cooldown** — after compacting channel X, don't re-evaluate it for + `cooldown_minutes`. Track in-memory: + ```go + lastCompacted map[string]time.Time // channelID → timestamp + ``` + +2. **In-flight dedup** — if channel X is currently being compacted (sem + acquired, goroutine running), skip it in the current scan. Track via: + ```go + inFlight sync.Map // channelID → struct{} + ``` + +3. **Stacking** — the existing summary boundary logic already handles + multiple summaries correctly. Auto-compaction produces the same tree + nodes as manual, so stacking "just works." + +--- + +## 7. Lifecycle + Wiring + +### main.go changes + +```go +// After role resolver and stores setup: +compactionSvc := compaction.NewService(stores, roleResolver) +compactionScanner := compaction.NewScanner(compactionSvc, stores, compaction.ScannerConfig{ + Enabled: getGlobalBool(stores, "auto_compaction_enabled", false), + Interval: time.Duration(getGlobalInt(stores, "auto_compaction_interval_minutes", 5)) * time.Minute, + Concurrency: getGlobalInt(stores, "auto_compaction_concurrency", 2), +}) +compactionScanner.Start() +defer compactionScanner.Stop() // drain in-flight on shutdown + +// Summarize handler now wraps compaction service: +summarize := handlers.NewSummarizeHandler(stores, roleResolver, compactionSvc) +``` + +Follows the same pattern as `kbIngester` with `defer kbIngester.Wait()`. + +### Config reload + +The scanner re-reads global settings at the start of each scan cycle. +No restart required to change thresholds, intervals, or the kill switch. +If `auto_compaction_enabled` flips to `false`, the scanner skips all +work on the next tick. + +--- + +## 8. Observability + +### Logging + +``` +🔍 compaction: scanning 50 candidates +🔍 compaction: channel abc123 qualifies (78% of 128K context, 42 messages post-summary) +📝 compaction: compacting channel abc123 for user xyz (trigger=auto) +✅ compaction: channel abc123 done (42 messages → 847 chars, model=claude-sonnet-4-5-20250929) +⏭ compaction: channel def456 skipped (cooldown, 12m remaining) +⏭ compaction: channel ghi789 skipped (rate limit) +``` + +### Audit log + +Auto-compaction events are logged to `audit_log`: + +```go +stores.Audit.Log(ctx, &models.AuditEntry{ + Actor: "system:compaction", + Action: "compaction.auto", + Resource: "channel:" + channelID, + Details: models.JSONMap{ + "summarized_count": result.SummarizedCount, + "model": result.Model, + "trigger": "auto", + }, +}) +``` + +### Usage tracking + +All auto-compaction calls flow through the existing usage logging path +(same as manual summarization). The `metadata.trigger = "auto"` field +lets the admin usage dashboard distinguish auto from manual. + +--- + +## 9. Frontend Changes + +Minimal — the backend does the heavy lifting. + +### Auto-compaction indicator + +When loading a conversation that was auto-compacted, the frontend +already handles summary boundaries correctly (the collapsed history +toggle, the "X earlier messages summarized" bar). The only new UI is +a subtle indicator on the summary node: + +``` +▸ 42 earlier messages summarized (auto) +``` + +The `(auto)` label comes from `metadata.trigger` on the summary message. + +### Channel settings + +Add compaction controls to the channel settings panel (Section 4 above). +Saves to `PATCH /channels/:id` with `settings` field — no new endpoint. + +### Admin panel + +Add the auto-compaction card to the AI settings section (Section 5). +Saves via existing `PUT /admin/settings/:key` endpoints. + +--- + +## 10. Migration + +**No new migration required.** + +- Channel compaction settings → existing `Channel.Settings` JSONB +- Admin settings → existing `global_settings` table +- Summary messages → existing `messages` table with metadata +- Usage tracking → existing `usage_log` table +- Audit logging → existing `audit_log` table + +--- + +## 11. File Manifest + +``` +server/ + treepath/ ← NEW package + path.go (extracted from handlers/tree.go) + summary.go (isSummaryMessage + helpers) + siblings.go (sibling queries) + cursor.go (cursor + sibling index) + compaction/ ← NEW package + compaction.go (Service, Compact, CompactRequest/Result) + scanner.go (Scanner, scan loop, candidate query) + estimator.go (EstimateTokens, EstimatePath) + handlers/ + tree.go (updated: delegates to treepath) + summarize.go (updated: delegates to compaction.Service) + completion.go (updated: imports treepath) + main.go (updated: wire compaction scanner) +src/js/ + ui-core.js (updated: show "(auto)" on auto-compacted summaries) + ui-settings.js (updated: channel compaction controls) + ui-admin.js (updated: auto-compaction admin card) +``` + +--- + +## 12. Implementation Order + +**Phase 1: Extract** — Move tree helpers to `treepath`, core +summarization to `compaction.Service`. Refactor `SummarizeHandler` to +delegate. All existing tests must still pass. No new behavior. + +**Phase 2: Estimator + Scanner** — Add `estimator.go`, `scanner.go`, +candidate query, `shouldCompact` logic. Wire into `main.go` with +`Start()`/`Stop()`. Global settings for admin controls. Ship with +`auto_compaction_enabled` defaulting to `false`. + +**Phase 3: Frontend** — Channel compaction settings, admin panel card, +auto-compaction indicator on summary nodes. + +**Phase 4: Soak + tune** — Enable on a test instance, watch logs, tune +default threshold and cooldown. Adjust candidate query filters if scan +is too expensive. + +--- + +## 13. What This Intentionally Doesn't Do + +- **Token-accurate estimation** — chars/4 is good enough. A real + tokenizer adds a dependency for marginal accuracy gain. Revisit if + users report summaries firing too early or too late. + +- **Per-model prompt tuning** — the summarization prompt is one-size-fits + -all. Works fine across Claude/GPT/Llama. Revisit if quality varies + significantly across providers. + +- **Group channel compaction** — auto-compaction targets `direct` channels + only (single-owner). Group/channel types need multi-cursor awareness + (which user's path to compact?). Deferred to v0.19.0 multi-participant. + +- **Compaction snapshots for KB** — the roadmap mentions this. The + summary nodes we create here are already stored as messages with full + metadata. KB integration can query them later without changes to this + design. + +- **Event-driven triggers** — a post-completion hook that checks + immediately after each message would reduce latency vs. the ticker. + Worth adding later but the ticker is simpler to ship and debug. The + 2-minute activity gap in the candidate query achieves a similar effect + (don't compact mid-conversation). diff --git a/EXTENSIONS.md b/docs/EXTENSIONS.md similarity index 100% rename from EXTENSIONS.md rename to docs/EXTENSIONS.md diff --git a/REFACTOR-0.10.3.md b/docs/REFACTOR-0.10.3.md similarity index 100% rename from REFACTOR-0.10.3.md rename to docs/REFACTOR-0.10.3.md diff --git a/ROADMAP.md b/docs/ROADMAP.md similarity index 100% rename from ROADMAP.md rename to docs/ROADMAP.md diff --git a/UI-AUDIT-0.10.5.md b/docs/UI-AUDIT-0.10.5.md similarity index 100% rename from UI-AUDIT-0.10.5.md rename to docs/UI-AUDIT-0.10.5.md diff --git a/server/compaction/compaction.go b/server/compaction/compaction.go new file mode 100644 index 0000000..6435728 --- /dev/null +++ b/server/compaction/compaction.go @@ -0,0 +1,312 @@ +package compaction + +import ( + "context" + "encoding/json" + "fmt" + "log" + "strings" + "time" + + "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/providers" + "git.gobha.me/xcaliber/chat-switchboard/roles" + "git.gobha.me/xcaliber/chat-switchboard/store" + "git.gobha.me/xcaliber/chat-switchboard/treepath" +) + +// ── Request / Result ──────────────────────── + +// CompactRequest specifies what to compact. +type CompactRequest struct { + ChannelID string + UserID string // channel owner + TeamID *string // for role resolution + Trigger string // "manual" | "auto" +} + +// CompactResult describes what compaction produced. +type CompactResult struct { + SummaryID string + SummarizedCount int + Model string + UsedFallback bool + Content string + InputTokens int + OutputTokens int +} + +// ── Errors ────────────────────────────────── + +var ( + // ErrContextBudget is returned when the conversation to summarize + // exceeds the utility model's context window. + ErrContextBudget = fmt.Errorf("conversation exceeds utility model context window") +) + +// ── Service ───────────────────────────────── + +// Service provides conversation compaction (summarization). +// Used by both the HTTP handler (manual) and the background scanner (auto). +type Service struct { + stores store.Stores + resolver *roles.Resolver +} + +// NewService creates a compaction service. +func NewService(stores store.Stores, resolver *roles.Resolver) *Service { + return &Service{stores: stores, resolver: resolver} +} + +// Resolver exposes the underlying role resolver (for external checks like +// IsConfigured / IsPersonalOverride). +func (s *Service) Resolver() *roles.Resolver { + return s.resolver +} + +// ── Compact ───────────────────────────────── + +// Compact summarizes the conversation in a channel, inserting a summary +// tree node. Returns the result or an error. +// +// The method: +// 1. Loads the active path via the message tree +// 2. Finds the most recent summary boundary +// 3. Collects post-boundary messages +// 4. Checks the content fits the utility model's context window +// 5. Calls the utility model role to generate a summary +// 6. Inserts the summary as a tree node with metadata +// 7. Updates the user's cursor +// 8. Logs usage +func (s *Service) Compact(ctx context.Context, req CompactRequest) (*CompactResult, error) { + // ── Load active path ── + path, err := treepath.GetActivePath(req.ChannelID, req.UserID) + if err != nil { + return nil, fmt.Errorf("load conversation: %w", err) + } + if len(path) < 4 { + return nil, fmt.Errorf("conversation too short to summarize") + } + + // Find existing summary boundary (if any) and only summarize after it + startIdx := 0 + for i := range path { + if treepath.IsSummaryMessage(&path[i]) { + startIdx = i + 1 + } + } + + // Build messages to summarize (skip system messages, skip previous summaries) + var toSummarize []string + messagesInScope := 0 + var lastMessageID string + for _, m := range path[startIdx:] { + if m.Role == "system" || treepath.IsSummaryMessage(&m) { + continue + } + toSummarize = append(toSummarize, fmt.Sprintf("%s: %s", m.Role, m.Content)) + messagesInScope++ + lastMessageID = m.ID + } + + if messagesInScope < 4 { + return nil, fmt.Errorf("not enough new messages since last summary") + } + + // ── Build the summarization prompt ── + conversationText := strings.Join(toSummarize, "\n\n") + systemPrompt := `You are a conversation summarizer. Create a concise but comprehensive summary of the following conversation that preserves: +- Key decisions and conclusions reached +- Important facts, names, numbers, and technical details mentioned +- Action items or commitments made +- The overall context and topic flow + +Write the summary in a way that would allow the conversation to continue naturally. Use clear, factual language. Do not include meta-commentary about the summarization process.` + + userContent := "Summarize this conversation:\n\n" + conversationText + + // ── Context budget guard rail ── + // Estimate whether the prompt fits the utility model's context window. + // Prevents sending truncated input to small models (e.g. 32K utility + // summarizing a 128K conversation). + promptTokens := EstimateTokens(systemPrompt) + 4 + EstimateTokens(userContent) + 4 + utilityBudget := s.getUtilityContextBudget(ctx, req.UserID, req.TeamID) + // Reserve 20% for output (the summary itself) + inputCeiling := int(float64(utilityBudget) * 0.80) + + if promptTokens > inputCeiling { + log.Printf("⚠ Compaction skipped for channel %s: prompt ~%dK tokens exceeds utility model budget %dK (ceiling %dK)", + req.ChannelID, promptTokens/1000, utilityBudget/1000, inputCeiling/1000) + return nil, fmt.Errorf("%w: ~%d tokens needed, utility model allows ~%d", + ErrContextBudget, promptTokens, inputCeiling) + } + + summaryPrompt := []providers.Message{ + {Role: "system", Content: systemPrompt}, + {Role: "user", Content: userContent}, + } + + // ── Call utility role ── + log.Printf("📝 Compacting channel %s for user %s (%d messages, ~%dK tokens, trigger=%s)", + req.ChannelID, req.UserID, messagesInScope, promptTokens/1000, req.Trigger) + + result, err := s.resolver.Complete(ctx, roles.RoleUtility, req.UserID, req.TeamID, summaryPrompt) + if err != nil { + log.Printf("⚠ Compaction failed for channel %s: %v", req.ChannelID, err) + return nil, fmt.Errorf("summarization failed: %w", err) + } + + // ── Log usage ── + s.logUsage(ctx, req.ChannelID, req.UserID, req.Trigger, result) + + // ── Insert summary message as a tree node ── + metadata := models.JSONMap{ + "type": "summary", + "summarized_until_id": lastMessageID, + "summarized_count": messagesInScope, + "utility_model": result.Model, + "used_fallback": result.UsedFallback, + "trigger": req.Trigger, + } + + metaJSON, _ := json.Marshal(metadata) + siblingIdx := treepath.NextSiblingIndex(req.ChannelID, &lastMessageID) + + var summaryMsgID string + err = database.DB.QueryRow(` + INSERT INTO messages (channel_id, parent_id, role, content, model, metadata, sibling_index, participant_type) + VALUES ($1, $2, 'assistant', $3, $4, $5, $6, 'system') + RETURNING id + `, req.ChannelID, lastMessageID, result.Content, result.Model, string(metaJSON), siblingIdx).Scan(&summaryMsgID) + if err != nil { + log.Printf("⚠ Failed to persist summary for channel %s: %v", req.ChannelID, err) + return nil, fmt.Errorf("failed to save summary: %w", err) + } + + // Update cursor to point to the summary node + if err := treepath.UpdateCursor(req.ChannelID, req.UserID, summaryMsgID); err != nil { + log.Printf("⚠ Failed to update cursor after compaction: %v", err) + } + + log.Printf("✅ Compaction done for channel %s: %s (%d messages → %d chars, trigger=%s)", + req.ChannelID, summaryMsgID, messagesInScope, len(result.Content), req.Trigger) + + return &CompactResult{ + SummaryID: summaryMsgID, + SummarizedCount: messagesInScope, + Model: result.Model, + UsedFallback: result.UsedFallback, + Content: result.Content, + InputTokens: result.InputTokens, + OutputTokens: result.OutputTokens, + }, nil +} + +// ── Context Budget ────────────────────────── + +// getUtilityContextBudget resolves the utility model's context window. +// Falls back to DefaultBudget if the model isn't in the catalog. +func (s *Service) getUtilityContextBudget(ctx context.Context, userID string, teamID *string) int { + cfg, err := s.resolver.GetConfig(ctx, roles.RoleUtility, userID, teamID) + if err != nil || cfg == nil || cfg.Primary == nil { + return DefaultBudget + } + + // Look up the primary model in catalog + entry, err := s.stores.Catalog.GetByModelID(ctx, cfg.Primary.ProviderConfigID, cfg.Primary.ModelID) + if err == nil && entry.Capabilities.MaxContext > 0 { + return entry.Capabilities.MaxContext + } + + // Try any-provider lookup (model may be synced under a different config) + entry, err = s.stores.Catalog.GetByModelIDAny(ctx, cfg.Primary.ModelID) + if err == nil && entry.Capabilities.MaxContext > 0 { + return entry.Capabilities.MaxContext + } + + return DefaultBudget +} + +// ── Rate Limit Check ──────────────────────── + +// CheckRateLimit checks the utility rate limit for org-funded calls. +// Returns nil if within limits, an error with a descriptive message otherwise. +func (s *Service) CheckRateLimit(ctx context.Context, userID string) error { + // Load rate limit from global settings (default: 20/hour, 0 = unlimited) + limit := 20 + settings, err := s.stores.GlobalConfig.Get(ctx, "utility_rate_limit") + if err == nil { + if v, ok := settings["value"]; ok { + switch n := v.(type) { + case float64: + limit = int(n) + case int: + limit = n + } + } + } + + if limit <= 0 { + return nil // unlimited + } + + since := time.Now().Add(-1 * time.Hour) + count, err := s.stores.Usage.CountRecentByRole(ctx, userID, roles.RoleUtility, since) + if err != nil { + log.Printf("⚠ Rate limit check failed: %v", err) + return nil // fail open — don't block on DB errors + } + + if count >= limit { + return fmt.Errorf("rate limit exceeded: %d utility calls in the last hour (limit: %d)", count, limit) + } + return nil +} + +// ── Helpers ───────────────────────────────── + +// GetUserTeamID returns the user's first team ID (for role resolution). +func (s *Service) GetUserTeamID(ctx context.Context, userID string) *string { + var teamID string + err := database.DB.QueryRow(` + SELECT team_id FROM team_members WHERE user_id = $1 LIMIT 1 + `, userID).Scan(&teamID) + if err != nil || teamID == "" { + return nil + } + return &teamID +} + +func (s *Service) logUsage(ctx context.Context, channelID, userID, trigger string, result *roles.CompletionResult) { + role := roles.RoleUtility + entry := &models.UsageEntry{ + ChannelID: &channelID, + UserID: userID, + ProviderConfigID: &result.ConfigID, + ProviderScope: result.ProviderScope, + ModelID: result.Model, + Role: &role, + PromptTokens: result.InputTokens, + CompletionTokens: result.OutputTokens, + CacheCreationTokens: result.CacheCreationTokens, + CacheReadTokens: result.CacheReadTokens, + } + + // Calculate cost from pricing + pricing, err := s.stores.Pricing.GetForModel(ctx, result.ConfigID, result.Model) + if err == nil && pricing != nil { + if pricing.InputPerM != nil { + costIn := float64(result.InputTokens) / 1_000_000 * *pricing.InputPerM + entry.CostInput = &costIn + } + if pricing.OutputPerM != nil { + costOut := float64(result.OutputTokens) / 1_000_000 * *pricing.OutputPerM + entry.CostOutput = &costOut + } + } + + if err := s.stores.Usage.Log(ctx, entry); err != nil { + log.Printf("⚠ Failed to log compaction usage: %v", err) + } +} diff --git a/server/compaction/compaction_test.go b/server/compaction/compaction_test.go new file mode 100644 index 0000000..628f536 --- /dev/null +++ b/server/compaction/compaction_test.go @@ -0,0 +1,401 @@ +package compaction + +import ( + "context" + "encoding/json" + "testing" + "time" + + "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/models" + postgres "git.gobha.me/xcaliber/chat-switchboard/store/postgres" +) + +// ── Helpers ───────────────────────────────── + +func requireDB(t *testing.T) { + t.Helper() + database.RequireTestDB(t) + database.TruncateAll(t) +} + +func seedChannel(t *testing.T, userID, title string, msgCount, contentSize int) (channelID string, msgIDs []string) { + t.Helper() + channelID = database.SeedTestChannel(t, userID, title) + msgIDs = database.SeedTestMessages(t, channelID, msgCount, contentSize) + if len(msgIDs) > 0 { + database.SeedTestCursor(t, channelID, userID, msgIDs[len(msgIDs)-1]) + } + return +} + +// setGlobalSetting writes a global_settings key for scanner config tests. +func setGlobalSetting(t *testing.T, key string, value interface{}) { + t.Helper() + valMap := models.JSONMap{"value": value} + valJSON, _ := json.Marshal(valMap) + _, err := database.DB.Exec(` + INSERT INTO global_settings (key, value) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET value = $2 + `, key, string(valJSON)) + if err != nil { + t.Fatalf("setGlobalSetting(%s): %v", key, err) + } +} + +// touchChannelTime backdates a channel's updated_at. +func touchChannelTime(t *testing.T, channelID string, age time.Duration) { + t.Helper() + target := time.Now().Add(-age) + _, err := database.DB.Exec(`UPDATE channels SET updated_at = $1 WHERE id = $2`, target, channelID) + if err != nil { + t.Fatalf("touchChannelTime: %v", err) + } +} + +// setChannelSettings sets the channel.settings JSONB. +func setChannelSettings(t *testing.T, channelID string, settings models.JSONMap) { + t.Helper() + sJSON, _ := json.Marshal(settings) + _, err := database.DB.Exec(`UPDATE channels SET settings = $1 WHERE id = $2`, string(sJSON), channelID) + if err != nil { + t.Fatalf("setChannelSettings: %v", err) + } +} + +// ═══════════════════════════════════════════ +// Estimator Tests (no DB — duplicated here for package-level access) +// ═══════════════════════════════════════════ +// Note: estimator_test.go covers the unit tests more thoroughly. +// These are here to verify the package compiles with DB tests. + +func TestEstimateTokens_Basic(t *testing.T) { + if got := EstimateTokens("hello world!"); got != 3 { // 12 chars → 3 + t.Errorf("EstimateTokens = %d, want 3", got) + } +} + +// ═══════════════════════════════════════════ +// Scanner: Candidate Query +// ═══════════════════════════════════════════ + +func TestScanner_FindCandidates_ReturnsQualifying(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) // no resolver needed for this test + sc := NewScanner(svc, stores, ScannerConfig{}) + + userID := database.SeedTestUser(t, "scanuser1", "scan1@test.com") + + // Create a channel with enough messages to qualify + // 20 messages × 2000 chars = 40K chars (> candidateMinChars=20K) + channelID, _ := seedChannel(t, userID, "Big Chat", 20, 2000) + + // Backdate so it passes the activity gap (>2min old) and recency (<7 days) + touchChannelTime(t, channelID, 5*time.Minute) + + ctx := context.Background() + candidates := sc.findCandidates(ctx) + + if len(candidates) == 0 { + t.Fatal("expected at least 1 candidate, got 0") + } + + found := false + for _, c := range candidates { + if c.ID == channelID { + found = true + break + } + } + if !found { + t.Fatalf("channel %s not in candidates (got %d candidates)", channelID, len(candidates)) + } +} + +func TestScanner_FindCandidates_ExcludesTooFewMessages(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + + userID := database.SeedTestUser(t, "scanuser2", "scan2@test.com") + + // Only 4 messages — below candidateMinMessages=10 + channelID, _ := seedChannel(t, userID, "Small Chat", 4, 2000) + touchChannelTime(t, channelID, 5*time.Minute) + + ctx := context.Background() + candidates := sc.findCandidates(ctx) + + for _, c := range candidates { + if c.ID == channelID { + t.Fatal("channel with <10 messages should not be a candidate") + } + } +} + +func TestScanner_FindCandidates_ExcludesTooRecent(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + + userID := database.SeedTestUser(t, "scanuser3", "scan3@test.com") + + // Enough messages but updated just now (< candidateActivityGap=2min) + seedChannel(t, userID, "Fresh Chat", 20, 2000) + // Don't backdate — should be excluded + + ctx := context.Background() + candidates := sc.findCandidates(ctx) + + for _, c := range candidates { + if c.UserID == userID { + t.Fatal("channel updated just now should not be a candidate (activity gap)") + } + } +} + +func TestScanner_FindCandidates_ExcludesArchived(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + + userID := database.SeedTestUser(t, "scanuser4", "scan4@test.com") + channelID, _ := seedChannel(t, userID, "Archived Chat", 20, 2000) + touchChannelTime(t, channelID, 5*time.Minute) + + // Archive the channel + database.DB.Exec(`UPDATE channels SET is_archived = true WHERE id = $1`, channelID) + + ctx := context.Background() + candidates := sc.findCandidates(ctx) + + for _, c := range candidates { + if c.ID == channelID { + t.Fatal("archived channel should not be a candidate") + } + } +} + +// ═══════════════════════════════════════════ +// Scanner: Cooldown + Dedup +// ═══════════════════════════════════════════ + +func TestScanner_Cooldown(t *testing.T) { + sc := &Scanner{ + lastCompacted: make(map[string]time.Time), + } + + channelID := "test-cooldown-channel" + + // No cooldown initially + sc.mu.Lock() + _, inCooldown := sc.lastCompacted[channelID] + sc.mu.Unlock() + if inCooldown { + t.Fatal("should not be in cooldown initially") + } + + // Record compaction + sc.mu.Lock() + sc.lastCompacted[channelID] = time.Now() + sc.mu.Unlock() + + // Now it's in cooldown + sc.mu.Lock() + last, ok := sc.lastCompacted[channelID] + sc.mu.Unlock() + if !ok { + t.Fatal("should be in cooldown after recording") + } + if time.Since(last) > time.Second { + t.Fatal("cooldown timestamp should be recent") + } +} + +func TestScanner_InFlightDedup(t *testing.T) { + sc := &Scanner{} + // sync.Map zero value is ready to use + + channelID := "test-dedup-channel" + + // First load — not in flight + _, loaded := sc.inFlight.LoadOrStore(channelID, struct{}{}) + if loaded { + t.Fatal("should not be loaded on first attempt") + } + + // Second load — already in flight + _, loaded = sc.inFlight.LoadOrStore(channelID, struct{}{}) + if !loaded { + t.Fatal("should be loaded on second attempt (dedup)") + } + + // Clean up + sc.inFlight.Delete(channelID) + _, loaded = sc.inFlight.LoadOrStore(channelID, struct{}{}) + if loaded { + t.Fatal("should not be loaded after delete") + } +} + +// ═══════════════════════════════════════════ +// Scanner: Channel Opt-Out +// ═══════════════════════════════════════════ + +func TestScanner_ShouldCompact_ChannelOptOut(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) // no resolver → IsConfigured returns false + sc := NewScanner(svc, stores, ScannerConfig{}) + + userID := database.SeedTestUser(t, "optout_user", "optout@test.com") + channelID, _ := seedChannel(t, userID, "Opt-Out Chat", 20, 2000) + + // Opt out via channel settings + setChannelSettings(t, channelID, models.JSONMap{"auto_compaction": false}) + + ch := models.Channel{Settings: models.JSONMap{"auto_compaction": false}} + ch.ID = channelID + ch.UserID = userID + + ctx := context.Background() + if sc.shouldCompact(ctx, &ch) { + t.Fatal("shouldCompact should return false for opted-out channel") + } +} + +// ═══════════════════════════════════════════ +// Scanner: Settings Helpers +// ═══════════════════════════════════════════ + +func TestScanner_IsEnabled(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + ctx := context.Background() + + // Default: disabled + if sc.isEnabled(ctx) { + t.Fatal("should be disabled by default") + } + + // Enable + setGlobalSetting(t, "auto_compaction_enabled", true) + if !sc.isEnabled(ctx) { + t.Fatal("should be enabled after setting to true") + } + + // Disable again + setGlobalSetting(t, "auto_compaction_enabled", false) + if sc.isEnabled(ctx) { + t.Fatal("should be disabled after setting to false") + } +} + +func TestScanner_GetThreshold_Default(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + + ch := &models.Channel{} + got := sc.getThreshold(ch) + if got != DefaultThreshold { + t.Errorf("default threshold = %f, want %f", got, DefaultThreshold) + } +} + +func TestScanner_GetThreshold_GlobalOverride(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + + setGlobalSetting(t, "auto_compaction_threshold", 0.85) + + ch := &models.Channel{} + got := sc.getThreshold(ch) + if got != 0.85 { + t.Errorf("global threshold = %f, want 0.85", got) + } +} + +func TestScanner_GetThreshold_ChannelOverride(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + + setGlobalSetting(t, "auto_compaction_threshold", 0.85) + + ch := &models.Channel{ + Settings: models.JSONMap{"compaction_threshold": 0.60}, + } + got := sc.getThreshold(ch) + if got != 0.60 { + t.Errorf("channel threshold = %f, want 0.60", got) + } +} + +func TestScanner_GetCooldownDuration_Default(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + ctx := context.Background() + + got := sc.getCooldownDuration(ctx) + if got != DefaultCooldown { + t.Errorf("default cooldown = %s, want %s", got, DefaultCooldown) + } +} + +func TestScanner_GetCooldownDuration_Override(t *testing.T) { + requireDB(t) + stores := postgres.NewStores(database.TestDB) + svc := NewService(stores, nil) + sc := NewScanner(svc, stores, ScannerConfig{}) + ctx := context.Background() + + setGlobalSetting(t, "auto_compaction_cooldown_minutes", float64(15)) + + got := sc.getCooldownDuration(ctx) + want := 15 * time.Minute + if got != want { + t.Errorf("cooldown = %s, want %s", got, want) + } +} + +// ═══════════════════════════════════════════ +// Context Budget Guard Rail +// ═══════════════════════════════════════════ + +func TestEstimateTokens_GuardRailMath(t *testing.T) { + // Simulate a large conversation that exceeds a 32K utility model + // 100K chars ≈ 25K tokens of conversation + system prompt overhead + largeContent := make([]byte, 100000) + contentTokens := EstimateTokens(string(largeContent)) + systemTokens := EstimateTokens("You are a conversation summarizer...") + 8 // +overhead + + totalPrompt := contentTokens + systemTokens + utilityBudget := 32000 // 32K model + inputCeiling := int(float64(utilityBudget) * 0.80) + + if totalPrompt <= inputCeiling { + t.Errorf("100K chars (%d tokens) should exceed 32K model ceiling (%d tokens)", + totalPrompt, inputCeiling) + } + + // Smaller conversation should fit + smallContent := make([]byte, 50000) + smallTokens := EstimateTokens(string(smallContent)) + systemTokens + if smallTokens > inputCeiling { + t.Errorf("50K chars (%d tokens) should fit in 32K model ceiling (%d tokens)", + smallTokens, inputCeiling) + } +} diff --git a/server/compaction/estimator.go b/server/compaction/estimator.go new file mode 100644 index 0000000..8bd7579 --- /dev/null +++ b/server/compaction/estimator.go @@ -0,0 +1,57 @@ +package compaction + +import ( + "git.gobha.me/xcaliber/chat-switchboard/treepath" +) + +// ── Token Estimation ──────────────────────── +// +// Rough heuristic matching the frontend Tokens.estimate() (~4 chars/token). +// Good enough for threshold checks. Swap in a real tokenizer later if +// over/under-triggering becomes a problem. + +// EstimateTokens returns a rough token count for a string. +// Uses the ~4 chars/token heuristic for English (GPT/Claude average). +func EstimateTokens(text string) int { + return (len(text) + 3) / 4 +} + +// EstimatePath returns total estimated tokens for a message path. +// Adds +4 per message for role/delimiter overhead, matching the frontend. +func EstimatePath(path []treepath.PathMessage) int { + total := 0 + for i := range path { + total += EstimateTokens(path[i].Content) + 4 + } + return total +} + +// EstimatePathFromSummary returns estimated tokens for messages after +// the most recent summary boundary. Returns 0 and false if a summary +// exists but fewer than minMessages follow it. +func EstimatePathFromSummary(path []treepath.PathMessage, minMessages int) (tokens int, msgCount int, ok bool) { + startIdx := 0 + for i := range path { + if treepath.IsSummaryMessage(&path[i]) { + startIdx = i + 1 + } + } + + post := path[startIdx:] + + // Count non-system, non-summary messages + count := 0 + est := 0 + for i := range post { + if post[i].Role == "system" || treepath.IsSummaryMessage(&post[i]) { + continue + } + count++ + est += EstimateTokens(post[i].Content) + 4 + } + + if count < minMessages { + return 0, count, false + } + return est, count, true +} diff --git a/server/compaction/estimator_test.go b/server/compaction/estimator_test.go new file mode 100644 index 0000000..20c5236 --- /dev/null +++ b/server/compaction/estimator_test.go @@ -0,0 +1,92 @@ +package compaction + +import ( + "testing" + + "git.gobha.me/xcaliber/chat-switchboard/treepath" +) + +func TestEstimateTokens(t *testing.T) { + tests := []struct { + input string + want int + }{ + {"", 0}, + {"a", 1}, + {"abcd", 1}, + {"abcde", 2}, + {"Hello, world!", 4}, // 13 chars → ceil(13/4) = 4 + {string(make([]byte, 100)), 25}, + {string(make([]byte, 1000)), 250}, + } + for _, tt := range tests { + got := EstimateTokens(tt.input) + if got != tt.want { + t.Errorf("EstimateTokens(%d chars) = %d, want %d", len(tt.input), got, tt.want) + } + } +} + +func TestEstimatePath(t *testing.T) { + path := []treepath.PathMessage{ + {Content: string(make([]byte, 40))}, // 10 tokens + 4 overhead = 14 + {Content: string(make([]byte, 80))}, // 20 tokens + 4 overhead = 24 + {Content: string(make([]byte, 120))}, // 30 tokens + 4 overhead = 34 + } + got := EstimatePath(path) + // (10+4) + (20+4) + (30+4) = 72 + if got != 72 { + t.Errorf("EstimatePath = %d, want 72", got) + } +} + +func TestEstimatePathFromSummary_NoSummary(t *testing.T) { + path := make([]treepath.PathMessage, 10) + for i := range path { + path[i] = treepath.PathMessage{Role: "user", Content: string(make([]byte, 40))} + } + + tokens, count, ok := EstimatePathFromSummary(path, 8) + if !ok { + t.Fatal("expected ok=true for 10 messages >= minMessages=8") + } + if count != 10 { + t.Errorf("count = %d, want 10", count) + } + // Each: 10 tokens + 4 overhead = 14, × 10 = 140 + if tokens != 140 { + t.Errorf("tokens = %d, want 140", tokens) + } +} + +func TestEstimatePathFromSummary_TooFew(t *testing.T) { + path := make([]treepath.PathMessage, 3) + for i := range path { + path[i] = treepath.PathMessage{Role: "user", Content: "hello"} + } + + _, _, ok := EstimatePathFromSummary(path, 8) + if ok { + t.Error("expected ok=false for 3 messages < minMessages=8") + } +} + +func TestEstimatePathFromSummary_SkipsSystemMessages(t *testing.T) { + path := []treepath.PathMessage{ + {Role: "system", Content: "You are helpful."}, + {Role: "user", Content: string(make([]byte, 40))}, + {Role: "assistant", Content: string(make([]byte, 40))}, + } + + tokens, count, ok := EstimatePathFromSummary(path, 1) + if !ok { + t.Fatal("expected ok=true") + } + // system skipped, 2 non-system messages + if count != 2 { + t.Errorf("count = %d, want 2", count) + } + if tokens != 28 { // 2 × (10 + 4) = 28 + t.Errorf("tokens = %d, want 28", tokens) + } +} diff --git a/server/compaction/scanner.go b/server/compaction/scanner.go new file mode 100644 index 0000000..8fe4f61 --- /dev/null +++ b/server/compaction/scanner.go @@ -0,0 +1,399 @@ +package compaction + +import ( + "context" + "encoding/json" + "log" + "sync" + "time" + + "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/roles" + "git.gobha.me/xcaliber/chat-switchboard/store" + "git.gobha.me/xcaliber/chat-switchboard/treepath" +) + +// ── Defaults ──────────────────────────────── + +const ( + DefaultInterval = 5 * time.Minute + DefaultConcurrency = 2 + DefaultThreshold = 0.70 + DefaultCooldown = 30 * time.Minute + DefaultBudget = 128000 // 128K tokens — conservative fallback + + // Candidate query filters + candidateMinMessages = 10 + candidateMinChars = 20000 // ~5K tokens + candidateActivityGap = 2 * time.Minute + candidateMaxAge = 7 * 24 * time.Hour + candidateBatchSize = 50 + + // shouldCompact requires this many post-summary messages + minPostSummaryMessages = 8 +) + +// ── Scanner Config ────────────────────────── + +// ScannerConfig holds startup configuration for the scanner. +// Settings are re-read from global_settings each tick so changes +// take effect without restart. +type ScannerConfig struct { + Interval time.Duration + Concurrency int +} + +// ── Scanner ───────────────────────────────── + +// Scanner runs a periodic background loop that finds channels exceeding +// their context threshold and auto-compacts them via the utility model role. +// +// Follows the Ingester pattern: goroutine pool with semaphore, WaitGroup +// for graceful shutdown. +type Scanner struct { + service *Service + stores store.Stores + + sem chan struct{} + wg sync.WaitGroup + stopCh chan struct{} + + // Cooldown: channelID → last compaction time + mu sync.Mutex + lastCompacted map[string]time.Time + + // In-flight dedup + inFlight sync.Map // channelID → struct{} + + interval time.Duration +} + +// NewScanner creates a compaction scanner. Call Start() to begin scanning. +func NewScanner(svc *Service, stores store.Stores, cfg ScannerConfig) *Scanner { + interval := cfg.Interval + if interval <= 0 { + interval = DefaultInterval + } + concurrency := cfg.Concurrency + if concurrency <= 0 { + concurrency = DefaultConcurrency + } + + return &Scanner{ + service: svc, + stores: stores, + sem: make(chan struct{}, concurrency), + stopCh: make(chan struct{}), + lastCompacted: make(map[string]time.Time), + interval: interval, + } +} + +// Start begins the scan loop in a background goroutine. +func (sc *Scanner) Start() { + sc.wg.Add(1) + go func() { + defer sc.wg.Done() + sc.loop() + }() + log.Printf("🔍 compaction scanner started (interval=%s)", sc.interval) +} + +// Stop signals the scanner to stop and waits for in-flight compactions +// to drain. Safe to call from main's defer chain. +func (sc *Scanner) Stop() { + close(sc.stopCh) + sc.wg.Wait() + log.Printf("🔍 compaction scanner stopped") +} + +// ── Main Loop ─────────────────────────────── + +func (sc *Scanner) loop() { + ticker := time.NewTicker(sc.interval) + defer ticker.Stop() + + for { + select { + case <-sc.stopCh: + return + case <-ticker.C: + sc.tick() + } + } +} + +func (sc *Scanner) tick() { + ctx := context.Background() + + // Re-read kill switch each tick (no restart required) + if !sc.isEnabled(ctx) { + return + } + + candidates := sc.findCandidates(ctx) + if len(candidates) == 0 { + return + } + + log.Printf("🔍 compaction: scanning %d candidates", len(candidates)) + + cooldown := sc.getCooldownDuration(ctx) + + for i := range candidates { + ch := candidates[i] + + // Skip if in-flight + if _, loaded := sc.inFlight.LoadOrStore(ch.ID, struct{}{}); loaded { + continue + } + + // Skip if in cooldown + sc.mu.Lock() + if last, ok := sc.lastCompacted[ch.ID]; ok && time.Since(last) < cooldown { + sc.mu.Unlock() + sc.inFlight.Delete(ch.ID) + remaining := cooldown - time.Since(last) + log.Printf("⏭ compaction: channel %s skipped (cooldown, %s remaining)", ch.ID, remaining.Round(time.Second)) + continue + } + sc.mu.Unlock() + + // Precise check (loads full path, estimates tokens) + if !sc.shouldCompact(ctx, &ch) { + sc.inFlight.Delete(ch.ID) + continue + } + + // Rate limit check (auto-compaction is always org-funded) + if err := sc.service.CheckRateLimit(ctx, ch.UserID); err != nil { + sc.inFlight.Delete(ch.ID) + log.Printf("⏭ compaction: channel %s skipped (rate limit)", ch.ID) + continue + } + + // Dispatch compaction + sc.wg.Add(1) + go func(ch models.Channel) { + defer sc.wg.Done() + defer sc.inFlight.Delete(ch.ID) + + // Acquire semaphore + sc.sem <- struct{}{} + defer func() { <-sc.sem }() + + sc.doCompact(ch) + }(ch) + } +} + +// ── Compact One Channel ───────────────────── + +func (sc *Scanner) doCompact(ch models.Channel) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + teamID := sc.service.GetUserTeamID(ctx, ch.UserID) + + result, err := sc.service.Compact(ctx, CompactRequest{ + ChannelID: ch.ID, + UserID: ch.UserID, + TeamID: teamID, + Trigger: "auto", + }) + if err != nil { + log.Printf("⚠ compaction: channel %s failed: %v", ch.ID, err) + return + } + + // Record cooldown + sc.mu.Lock() + sc.lastCompacted[ch.ID] = time.Now() + sc.mu.Unlock() + + // Audit log + sc.stores.Audit.Log(ctx, &models.AuditEntry{ + ActorID: nil, // system action + Action: "compaction.auto", + ResourceType: "channel", + ResourceID: ch.ID, + Metadata: models.JSONMap{ + "summarized_count": result.SummarizedCount, + "model": result.Model, + "trigger": "auto", + "user_id": ch.UserID, + }, + }) + + log.Printf("✅ compaction: channel %s done (%d messages → %d chars, model=%s)", + ch.ID, result.SummarizedCount, len(result.Content), result.Model) +} + +// ── Candidate Query ───────────────────────── + +func (sc *Scanner) findCandidates(ctx context.Context) []models.Channel { + activityGap := time.Now().Add(-candidateActivityGap) + maxAge := time.Now().Add(-candidateMaxAge) + + rows, err := database.DB.QueryContext(ctx, ` + SELECT c.id, c.user_id, c.model, c.settings::text, + COUNT(m.id) AS msg_count, + COALESCE(SUM(LENGTH(m.content)), 0) AS total_chars + FROM channels c + JOIN messages m ON m.channel_id = c.id AND m.deleted_at IS NULL + WHERE c.deleted_at IS NULL + AND c.type = 'direct' + AND c.is_archived = false + AND c.updated_at < $1 + AND c.updated_at > $2 + GROUP BY c.id + HAVING COUNT(m.id) >= $3 + AND COALESCE(SUM(LENGTH(m.content)), 0) > $4 + ORDER BY c.updated_at DESC + LIMIT $5 + `, activityGap, + maxAge, + candidateMinMessages, + candidateMinChars, + candidateBatchSize, + ) + if err != nil { + log.Printf("⚠ compaction: candidate query failed: %v", err) + return nil + } + defer rows.Close() + + var candidates []models.Channel + for rows.Next() { + var ch models.Channel + var settingsRaw string + var msgCount, totalChars int + if err := rows.Scan(&ch.ID, &ch.UserID, &ch.Model, &settingsRaw, &msgCount, &totalChars); err != nil { + log.Printf("⚠ compaction: candidate scan: %v", err) + continue + } + if settingsRaw != "" { + _ = json.Unmarshal([]byte(settingsRaw), &ch.Settings) + } + candidates = append(candidates, ch) + } + + return candidates +} + +// ── Precise Check ─────────────────────────── + +func (sc *Scanner) shouldCompact(ctx context.Context, ch *models.Channel) bool { + // 1. Channel-level opt-out + if ch.Settings != nil { + if v, ok := ch.Settings["auto_compaction"]; ok { + if b, ok := v.(bool); ok && !b { + return false + } + } + } + + // 2. Check utility role is configured (global level — personal overrides + // aren't checked because auto-compaction is org-funded) + if !sc.service.Resolver().IsConfigured(ctx, roles.RoleUtility) { + return false + } + + // 3. Load active path + path, err := treepath.GetActivePath(ch.ID, ch.UserID) + if err != nil || len(path) < candidateMinMessages { + return false + } + + // 4. Estimate tokens from post-summary messages + tokens, msgCount, ok := EstimatePathFromSummary(path, minPostSummaryMessages) + if !ok { + return false + } + + // 5. Compare against threshold + budget := sc.getContextBudget(ctx, ch) + threshold := sc.getThreshold(ch) + ratio := float64(tokens) / float64(budget) + + if ratio >= threshold { + log.Printf("🔍 compaction: channel %s qualifies (%.0f%% of %dK context, %d messages post-summary)", + ch.ID, ratio*100, budget/1000, msgCount) + return true + } + return false +} + +// ── Settings Helpers ──────────────────────── + +func (sc *Scanner) isEnabled(ctx context.Context) bool { + val, err := sc.stores.GlobalConfig.Get(ctx, "auto_compaction_enabled") + if err != nil || val == nil { + return false // default: off + } + if v, ok := val["value"]; ok { + switch b := v.(type) { + case bool: + return b + case string: + return b == "true" + } + } + return false +} + +func (sc *Scanner) getContextBudget(ctx context.Context, ch *models.Channel) int { + // Try channel's model from catalog + if ch.Model != "" { + entry, err := sc.stores.Catalog.GetByModelIDAny(ctx, ch.Model) + if err == nil && entry.Capabilities.MaxContext > 0 { + return entry.Capabilities.MaxContext + } + } + return DefaultBudget +} + +func (sc *Scanner) getThreshold(ch *models.Channel) float64 { + // Channel override + if ch.Settings != nil { + if v, ok := ch.Settings["compaction_threshold"]; ok { + switch f := v.(type) { + case float64: + if f > 0 && f < 1 { + return f + } + } + } + } + // Global override + ctx := context.Background() + val, err := sc.stores.GlobalConfig.Get(ctx, "auto_compaction_threshold") + if err == nil && val != nil { + if v, ok := val["value"]; ok { + if f, ok := v.(float64); ok && f > 0 && f < 1 { + return f + } + } + } + return DefaultThreshold +} + +func (sc *Scanner) getCooldownDuration(ctx context.Context) time.Duration { + val, err := sc.stores.GlobalConfig.Get(ctx, "auto_compaction_cooldown_minutes") + if err == nil && val != nil { + if v, ok := val["value"]; ok { + switch n := v.(type) { + case float64: + if n > 0 { + return time.Duration(n) * time.Minute + } + case int: + if n > 0 { + return time.Duration(n) * time.Minute + } + } + } + } + return DefaultCooldown +} diff --git a/server/compaction/testmain_test.go b/server/compaction/testmain_test.go new file mode 100644 index 0000000..9a52e15 --- /dev/null +++ b/server/compaction/testmain_test.go @@ -0,0 +1,15 @@ +package compaction + +import ( + "os" + "testing" + + "git.gobha.me/xcaliber/chat-switchboard/database" +) + +func TestMain(m *testing.M) { + teardown := database.SetupTestDB() + code := m.Run() + teardown() + os.Exit(code) +} diff --git a/server/database/seed_helpers.go b/server/database/seed_helpers.go new file mode 100644 index 0000000..599f105 --- /dev/null +++ b/server/database/seed_helpers.go @@ -0,0 +1,62 @@ +package database + +import ( + "strings" + "testing" +) + +// ── Additional Test Seed Helpers (v0.15.0) ── + +// SeedTestMessage creates a single message in a channel and returns the message ID. +func SeedTestMessage(t *testing.T, channelID, parentID, role, content string) string { + t.Helper() + var id string + var parentPtr *string + if parentID != "" { + parentPtr = &parentID + } + err := DB.QueryRow(` + INSERT INTO messages (channel_id, parent_id, role, content, sibling_index) + VALUES ($1, $2, $3, $4, 0) + RETURNING id + `, channelID, parentPtr, role, content).Scan(&id) + if err != nil { + t.Fatalf("SeedTestMessage: %v", err) + } + return id +} + +// SeedTestMessages creates a linear chain of alternating user/assistant messages. +// Returns all message IDs in order. The first message has no parent. +func SeedTestMessages(t *testing.T, channelID string, count int, contentSize int) []string { + t.Helper() + content := strings.Repeat("x", contentSize) + ids := make([]string, 0, count) + + parentID := "" + for i := 0; i < count; i++ { + role := "user" + if i%2 == 1 { + role = "assistant" + } + id := SeedTestMessage(t, channelID, parentID, role, content) + ids = append(ids, id) + parentID = id + } + + return ids +} + +// SeedTestCursor sets the active leaf for a user in a channel. +func SeedTestCursor(t *testing.T, channelID, userID, leafID string) { + t.Helper() + _, err := DB.Exec(` + INSERT INTO channel_cursors (channel_id, user_id, active_leaf_id) + VALUES ($1, $2, $3) + ON CONFLICT (channel_id, user_id) + DO UPDATE SET active_leaf_id = $3, updated_at = NOW() + `, channelID, userID, leafID) + if err != nil { + t.Fatalf("SeedTestCursor: %v", err) + } +} diff --git a/server/handlers/admin.go b/server/handlers/admin.go index c3a4b33..ea930af 100644 --- a/server/handlers/admin.go +++ b/server/handlers/admin.go @@ -163,27 +163,13 @@ func (h *AdminHandler) destroyVault(c *gin.Context, userID string) { return } - // Null out vault columns — user gets a fresh vault on next login - _, err := database.DB.Exec(` - UPDATE users - SET encrypted_uek = NULL, uek_salt = NULL, uek_nonce = NULL, vault_set = false - WHERE id = $1 - `, userID) - if err != nil { - log.Printf("⚠ Failed to clear vault for user %s: %v", userID, err) - } + deleted := DestroyVaultDB(c.Request.Context(), userID) // Evict from session cache (if user is currently logged in) h.uekCache.Evict(userID) - // Delete personal provider configs — their keys are undecryptable now - result, err := database.DB.Exec(` - DELETE FROM provider_configs WHERE scope = 'personal' AND owner_id = $1 - `, userID) - if err != nil { - log.Printf("⚠ Failed to delete personal providers for user %s: %v", userID, err) - } else if rows, _ := result.RowsAffected(); rows > 0 { - log.Printf(" 🔐 Destroyed %d personal provider(s) for user %s", rows, userID) + if deleted > 0 { + log.Printf(" 🔐 Destroyed %d personal provider(s) for user %s", deleted, userID) } h.auditLog(c, "user.vault_destroyed", "user", userID, models.JSONMap{ @@ -191,6 +177,40 @@ func (h *AdminHandler) destroyVault(c *gin.Context, userID string) { }) } +// ResetVault explicitly resets a user's vault without changing their password. +// Clears vault columns, deletes personal provider configs, evicts UEK cache. +// The user gets a fresh vault on their next login. +// POST /api/v1/admin/users/:id/vault/reset +func (h *AdminHandler) ResetVault(c *gin.Context) { + userID := c.Param("id") + + // Verify user exists + user, err := h.stores.Users.GetByID(c.Request.Context(), userID) + if err != nil || user == nil { + c.JSON(http.StatusNotFound, gin.H{"error": "user not found"}) + return + } + + if h.uekCache == nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "vault not configured"}) + return + } + + deleted := DestroyVaultDB(c.Request.Context(), userID) + h.uekCache.Evict(userID) + + h.auditLog(c, "user.vault_reset", "user", userID, models.JSONMap{ + "reason": "admin_manual_reset", + "providers_deleted": deleted, + }) + + log.Printf(" 🔐 Vault reset for user %s (%d personal provider(s) cleared)", userID, deleted) + c.JSON(http.StatusOK, gin.H{ + "message": "vault reset — user will get a fresh vault on next login", + "providers_deleted": deleted, + }) +} + func (h *AdminHandler) DeleteUser(c *gin.Context) { id := c.Param("id") if err := h.stores.Users.Delete(c.Request.Context(), id); err != nil { diff --git a/server/handlers/auth.go b/server/handlers/auth.go index 42cb5c3..0aa3d08 100644 --- a/server/handlers/auth.go +++ b/server/handlers/auth.go @@ -258,6 +258,68 @@ func hashToken(token string) string { // ── Vault Lifecycle ───────────────────────── +// DestroyVaultDB nullifies a user's vault columns and deletes personal +// provider configs. This is the shared DB-level operation used by: +// - unlockVault (stale seal recovery) +// - BootstrapAdmin / SeedUsers (password rotation at startup) +// - AdminHandler.destroyVault (admin-initiated reset) +// +// Does NOT evict from UEK cache or write audit logs — callers handle that. +func DestroyVaultDB(ctx context.Context, userID string) (providersDeleted int64) { + _, err := database.DB.ExecContext(ctx, ` + UPDATE users + SET encrypted_uek = NULL, uek_salt = NULL, uek_nonce = NULL, vault_set = false + WHERE id = $1 + `, userID) + if err != nil { + log.Printf("⚠ DestroyVaultDB: failed to clear vault columns for user %s: %v", userID, err) + } + + result, err := database.DB.ExecContext(ctx, ` + DELETE FROM provider_configs WHERE scope = 'personal' AND owner_id = $1 + `, userID) + if err != nil { + log.Printf("⚠ DestroyVaultDB: failed to delete personal providers for user %s: %v", userID, err) + return 0 + } + rows, _ := result.RowsAffected() + return rows +} + +// ProbeAndRepairVault checks whether a user's vault can be unlocked with +// the given password. If the vault doesn't exist (vault_set=false), this is +// a no-op. If the vault exists but the seal is stale (encrypted_uek was +// wrapped with a different password), the vault and personal providers are +// destroyed so initVault fires cleanly on next login. +// +// Used by BootstrapAdmin and SeedUsers where the password is known at +// startup but the UEK cache is not available. +func ProbeAndRepairVault(ctx context.Context, userID, password string) { + var vaultSet bool + var encryptedUEK, salt, nonce []byte + + err := database.DB.QueryRowContext(ctx, ` + SELECT vault_set, encrypted_uek, uek_salt, uek_nonce + FROM users WHERE id = $1 + `, userID).Scan(&vaultSet, &encryptedUEK, &salt, &nonce) + if err != nil || !vaultSet { + return // no vault to probe + } + + pdk := crypto.DeriveKeyFromPassword(password, salt) + if _, err := crypto.UnwrapUEK(encryptedUEK, nonce, pdk); err == nil { + return // vault seal matches current password — all good + } + + // Stale seal: password has actually changed since the vault was sealed + deleted := DestroyVaultDB(ctx, userID) + if deleted > 0 { + log.Printf(" 🔐 Vault stale-seal repair for user %s: cleared %d personal provider(s)", userID, deleted) + } else { + log.Printf(" 🔐 Vault stale-seal repair for user %s: vault columns cleared", userID) + } +} + // initVault generates a UEK, wraps it with the user's password, and stores // the encrypted UEK + salt + nonce on the user record. Called on registration // and on first login for pre-migration users. @@ -328,7 +390,20 @@ func (h *AuthHandler) unlockVault(ctx context.Context, user *models.User, passwo pdk := crypto.DeriveKeyFromPassword(password, salt) uek, err := crypto.UnwrapUEK(encryptedUEK, nonce, pdk) if err != nil { - log.Printf("⚠ Vault unlock failed for user %s: %v", user.ID, err) + // Stale seal: encrypted_uek was wrapped with a different password + // (e.g. admin reset, BootstrapAdmin rotation, or pre-UEK migration). + // The old UEK is irrecoverable. Destroy the vault and re-initialize + // with the current password so the user isn't permanently locked out. + deleted := DestroyVaultDB(ctx, user.ID) + if deleted > 0 { + log.Printf("⚠ Vault stale-seal recovery for user %s: cleared %d personal provider(s)", user.ID, deleted) + } else { + log.Printf("⚠ Vault stale-seal recovery for user %s: no personal providers to clear", user.ID) + } + + if err := h.initVault(ctx, user.ID, password); err != nil { + log.Printf("⚠ Vault re-init failed for user %s: %v", user.ID, err) + } return } @@ -356,6 +431,9 @@ func BootstrapAdmin(cfg *config.Config, s store.Stores) { "role": models.UserRoleAdmin, "is_active": true, }) + // If the actual password changed, the vault seal is stale. Probe it + // with the current password and destroy only if unwrap fails. + ProbeAndRepairVault(ctx, existing.ID, cfg.AdminPassword) log.Printf(" ✅ Admin user '%s' updated", cfg.AdminUsername) return } @@ -437,6 +515,8 @@ func SeedUsers(cfg *config.Config, s store.Stores) { "role": role, "is_active": true, }) + // Probe vault with current password — only destroys if seal is stale + ProbeAndRepairVault(ctx, existing.ID, password) log.Printf(" 🌱 Seed user '%s' updated (role=%s)", username, role) continue } diff --git a/server/handlers/live_compaction_test.go b/server/handlers/live_compaction_test.go new file mode 100644 index 0000000..9a15ea8 --- /dev/null +++ b/server/handlers/live_compaction_test.go @@ -0,0 +1,257 @@ +package handlers + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "testing" + + "git.gobha.me/xcaliber/chat-switchboard/compaction" + "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/roles" + postgres "git.gobha.me/xcaliber/chat-switchboard/store/postgres" + "git.gobha.me/xcaliber/chat-switchboard/treepath" +) + +// ═══════════════════════════════════════════ +// Live Compaction Tests +// ═══════════════════════════════════════════ +// Requires: TEST_DATABASE_URL + VENICE_API_KEY +// +// Tests the full Compact() pipeline end-to-end: +// seed messages → call utility model → verify summary tree node +// ═══════════════════════════════════════════ + +func TestLive_CompactionFullPipeline(t *testing.T) { + h := setupHarness(t) + veniceKey := requireVeniceKey(t) + userID, adminToken := h.createAdminUser("compactadmin", "compact@test.com") + + // Set up Venice provider + enable qwen3-4b + configID, _ := setupVeniceWithModel(t, h, adminToken, veniceKey, veniceTestModel) + + // Configure utility role → qwen3-4b + w := h.request("PUT", "/api/v1/admin/roles/utility", adminToken, map[string]interface{}{ + "primary": map[string]interface{}{ + "provider_config_id": configID, + "model_id": veniceTestModel, + }, + }) + if w.Code != http.StatusOK { + t.Fatalf("configure utility role: %d: %s", w.Code, w.Body.String()) + } + t.Log(" ✓ Utility role configured with", veniceTestModel) + + // Create a channel with enough messages to summarize + w = h.request("POST", "/api/v1/channels", adminToken, map[string]interface{}{ + "title": "Compaction Test", "type": "direct", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create channel: %d: %s", w.Code, w.Body.String()) + } + var ch map[string]interface{} + decode(w, &ch) + channelID := ch["id"].(string) + + // Seed messages manually (not via completion — faster and cheaper) + msgs := []struct{ role, content string }{ + {"user", "I'm planning a trip to Japan next month. What cities should I visit?"}, + {"assistant", "For a Japan trip, I'd recommend Tokyo for modern culture, Kyoto for temples and traditional culture, Osaka for food, and Hiroshima for history. How long is your trip?"}, + {"user", "About 10 days. I'm most interested in food and temples."}, + {"assistant", "With 10 days focusing on food and temples, I'd suggest: 3 days in Kyoto for Fushimi Inari, Kinkaku-ji, and Arashiyama. 2 days in Osaka for Dotonbori street food, Kuromon Market, and Namba. 2 days in Tokyo for Tsukiji Outer Market, Senso-ji, and Meiji Shrine. Then day trips to Nara (deer park and Todai-ji) and Kamakura (Great Buddha)."}, + {"user", "That sounds great! What about budget? I'm trying to keep it under $3000 for the whole trip."}, + {"assistant", "A $3000 budget for 10 days in Japan is doable. Rough breakdown: Flights ~$800-1000, Accommodation ~$50-80/night in hostels or budget hotels ($500-800), JR Pass 14-day ~$380, Food ~$30-50/day ($300-500), Activities and temples ~$200. Total: $2180-2880. Tips: eat at konbini (convenience stores) for cheap meals, visit free shrines, and book accommodation early."}, + {"user", "Should I get a JR Pass or just buy individual tickets?"}, + {"assistant", "For your itinerary covering Tokyo, Kyoto, Osaka, Nara, and Kamakura, a 7-day JR Pass ($200) would cover most of your intercity travel. The Tokyo-Kyoto shinkansen alone costs $120 one-way. I'd recommend the 7-day pass starting when you leave Tokyo for Kyoto, then use IC cards (Suica/Pasmo) for local transit."}, + {"user", "Perfect. One more question — what's the best time to visit temples to avoid crowds?"}, + {"assistant", "Early morning is best for temples. Fushimi Inari at sunrise (5-6am) is magical and nearly empty. Kinkaku-ji opens at 9am — arrive right at opening. Arashiyama bamboo grove is best before 8am. For Senso-ji in Tokyo, go at dawn for beautiful photos. Weekdays are always less crowded than weekends."}, + } + + var lastMsgID string + for _, m := range msgs { + var parentPtr *string + if lastMsgID != "" { + parentPtr = &lastMsgID + } + err := database.TestDB.QueryRow(` + INSERT INTO messages (channel_id, parent_id, role, content, sibling_index) + VALUES ($1, $2, $3, $4, 0) + RETURNING id + `, channelID, parentPtr, m.role, m.content).Scan(&lastMsgID) + if err != nil { + t.Fatalf("seed message: %v", err) + } + } + + // Set cursor to last message + database.TestDB.Exec(` + INSERT INTO channel_cursors (channel_id, user_id, active_leaf_id) + VALUES ($1, $2, $3) + ON CONFLICT (channel_id, user_id) DO UPDATE SET active_leaf_id = $3 + `, channelID, userID, lastMsgID) + + t.Logf(" ✓ Seeded %d messages in channel %s", len(msgs), channelID) + + // ── Run compaction ── + stores := postgres.NewStores(database.TestDB) + resolver := roles.NewResolver(stores, nil) + svc := compaction.NewService(stores, resolver) + + result, err := svc.Compact(context.Background(), compaction.CompactRequest{ + ChannelID: channelID, + UserID: userID, + TeamID: nil, + Trigger: "manual", + }) + if err != nil { + t.Fatalf("Compact() failed: %v", err) + } + + t.Logf(" ✓ Compact result: %d messages summarized, model=%s, %d chars", + result.SummarizedCount, result.Model, len(result.Content)) + + // Verify summary was created + if result.SummarizedCount != len(msgs) { + t.Errorf("summarized_count = %d, want %d", result.SummarizedCount, len(msgs)) + } + if result.Content == "" { + t.Fatal("summary content should not be empty") + } + if result.Model != veniceTestModel { + t.Errorf("model = %q, want %q", result.Model, veniceTestModel) + } + + // Verify summary message exists in the tree + var summaryContent, summaryRole string + var metadataRaw []byte + err = database.TestDB.QueryRow(` + SELECT role, content, metadata FROM messages WHERE id = $1 + `, result.SummaryID).Scan(&summaryRole, &summaryContent, &metadataRaw) + if err != nil { + t.Fatalf("query summary message: %v", err) + } + if summaryRole != "assistant" { + t.Errorf("summary role = %q, want assistant", summaryRole) + } + + var metadata models.JSONMap + json.Unmarshal(metadataRaw, &metadata) + if metadata["type"] != "summary" { + t.Errorf("metadata.type = %q, want summary", metadata["type"]) + } + if metadata["trigger"] != "manual" { + t.Errorf("metadata.trigger = %q, want manual", metadata["trigger"]) + } + t.Logf(" ✓ Summary message verified: id=%s, metadata=%v", result.SummaryID, metadata) + + // Verify cursor was updated to point to summary + path, err := treepath.GetActivePath(channelID, userID) + if err != nil { + t.Fatalf("GetActivePath after compaction: %v", err) + } + + // The last message in the path should be the summary + if len(path) == 0 { + t.Fatal("path should not be empty after compaction") + } + lastPath := path[len(path)-1] + if lastPath.ID != result.SummaryID { + t.Errorf("active path leaf = %s, want summary %s", lastPath.ID, result.SummaryID) + } + t.Logf(" ✓ Cursor updated: path has %d messages, leaf is summary", len(path)) + + // Verify usage was logged + var usageCount int + database.TestDB.QueryRow(` + SELECT COUNT(*) FROM usage_log WHERE channel_id = $1 AND role = 'utility' + `, channelID).Scan(&usageCount) + if usageCount == 0 { + t.Error("expected usage_log entry for utility role") + } + t.Logf(" ✓ Usage logged: %d entries", usageCount) +} + +// TestLive_CompactionContextBudgetGuardRail verifies that Compact() +// rejects input that exceeds the utility model's context window. +func TestLive_CompactionContextBudgetGuardRail(t *testing.T) { + h := setupHarness(t) + veniceKey := requireVeniceKey(t) + _, adminToken := h.createAdminUser("guardrail_admin", "guardrail@test.com") + userID := database.SeedTestUser(t, "guardrail_user", "gruser@test.com") + + // Set up Venice + qwen3-4b (32K context) + configID, catalogEntryID := setupVeniceWithModel(t, h, adminToken, veniceKey, veniceTestModel) + + // Set max_context to 32000 in catalog (in case sync didn't populate it) + database.TestDB.Exec(` + UPDATE model_catalog SET capabilities = capabilities || '{"max_context": 32000}'::jsonb + WHERE id = $1 + `, catalogEntryID) + + // Configure utility role + h.request("PUT", "/api/v1/admin/roles/utility", adminToken, map[string]interface{}{ + "primary": map[string]interface{}{ + "provider_config_id": configID, + "model_id": veniceTestModel, + }, + }) + + // Create channel with huge messages (>32K context worth) + w := h.request("POST", "/api/v1/channels", adminToken, map[string]interface{}{ + "title": "Guard Rail Test", "type": "direct", + }) + var ch map[string]interface{} + decode(w, &ch) + channelID := ch["id"].(string) + + // Seed 20 messages × 8000 chars each = 160K chars ≈ 40K tokens + // This exceeds 32K × 0.80 = 25.6K token ceiling + bigContent := make([]byte, 8000) + for i := range bigContent { + bigContent[i] = 'a' + } + var lastMsgID string + for i := 0; i < 20; i++ { + var parentPtr *string + if lastMsgID != "" { + parentPtr = &lastMsgID + } + role := "user" + if i%2 == 1 { + role = "assistant" + } + database.TestDB.QueryRow(` + INSERT INTO messages (channel_id, parent_id, role, content, sibling_index) + VALUES ($1, $2, $3, $4, 0) RETURNING id + `, channelID, parentPtr, role, string(bigContent)).Scan(&lastMsgID) + } + database.TestDB.Exec(` + INSERT INTO channel_cursors (channel_id, user_id, active_leaf_id) + VALUES ($1, $2, $3) + ON CONFLICT (channel_id, user_id) DO UPDATE SET active_leaf_id = $3 + `, channelID, userID, lastMsgID) + + // Run compaction — should fail with context budget error + stores := postgres.NewStores(database.TestDB) + resolver := roles.NewResolver(stores, nil) + svc := compaction.NewService(stores, resolver) + + _, err := svc.Compact(context.Background(), compaction.CompactRequest{ + ChannelID: channelID, + UserID: userID, + TeamID: nil, + Trigger: "auto", + }) + if err == nil { + t.Fatal("Compact() should fail when content exceeds utility model context window") + } + + t.Logf(" ✓ Guard rail triggered: %v", err) + + // Verify it's specifically a context budget error + if !strings.Contains(err.Error(), "context window") { + t.Errorf("expected context window error, got: %v", err) + } +} diff --git a/server/handlers/summarize.go b/server/handlers/summarize.go index af15033..240de46 100644 --- a/server/handlers/summarize.go +++ b/server/handlers/summarize.go @@ -1,40 +1,30 @@ package handlers import ( - "context" - "encoding/json" - "fmt" - "log" "net/http" - "strings" - "time" "github.com/gin-gonic/gin" + "git.gobha.me/xcaliber/chat-switchboard/compaction" "git.gobha.me/xcaliber/chat-switchboard/database" - "git.gobha.me/xcaliber/chat-switchboard/models" - "git.gobha.me/xcaliber/chat-switchboard/providers" "git.gobha.me/xcaliber/chat-switchboard/roles" - "git.gobha.me/xcaliber/chat-switchboard/store" ) -// SummarizeHandler handles conversation summarization using the utility model role. +// SummarizeHandler handles manual conversation summarization via HTTP. type SummarizeHandler struct { - stores store.Stores - resolver *roles.Resolver + compaction *compaction.Service } -// NewSummarizeHandler creates a new handler. -func NewSummarizeHandler(s store.Stores, resolver *roles.Resolver) *SummarizeHandler { - return &SummarizeHandler{stores: s, resolver: resolver} +// NewSummarizeHandler creates a new handler backed by the compaction service. +func NewSummarizeHandler(svc *compaction.Service) *SummarizeHandler { + return &SummarizeHandler{compaction: svc} } // ── Summarize & Continue ────────────────── // POST /channels/:id/summarize // -// Calls the utility role to summarize the conversation history, inserts -// the summary as a special message node in the tree, and returns it. -// Subsequent completions will use the summary as context boundary. +// User-triggered compaction: calls the utility role to summarize the +// conversation history, inserts the summary as a tree node, and returns it. func (h *SummarizeHandler) Summarize(c *gin.Context) { channelID := c.Param("id") @@ -55,9 +45,9 @@ func (h *SummarizeHandler) Summarize(c *gin.Context) { } // ── Check utility role is configured ── - if !h.resolver.IsConfigured(c.Request.Context(), roles.RoleUtility) { - // If user doesn't have a personal override either, it's not available - if !h.resolver.IsPersonalOverride(c.Request.Context(), userID, roles.RoleUtility) { + resolver := h.compaction.Resolver() + if !resolver.IsConfigured(c.Request.Context(), roles.RoleUtility) { + if !resolver.IsPersonalOverride(c.Request.Context(), userID, roles.RoleUtility) { c.JSON(http.StatusServiceUnavailable, gin.H{ "error": "Utility model role is not configured. Ask your admin to set one, or configure your own in Settings → Model Roles.", }) @@ -66,205 +56,40 @@ func (h *SummarizeHandler) Summarize(c *gin.Context) { } // ── Rate limiting (org-funded calls only) ── - isPersonal := h.resolver.IsPersonalOverride(c.Request.Context(), userID, roles.RoleUtility) + isPersonal := resolver.IsPersonalOverride(c.Request.Context(), userID, roles.RoleUtility) if !isPersonal { - if err := h.checkRateLimit(c.Request.Context(), userID); err != nil { + if err := h.compaction.CheckRateLimit(c.Request.Context(), userID); err != nil { c.JSON(http.StatusTooManyRequests, gin.H{"error": err.Error()}) return } } - // ── Load active path ── - path, err := getActivePath(channelID, userID) + // ── Compact ── + teamID := h.compaction.GetUserTeamID(c.Request.Context(), userID) + result, err := h.compaction.Compact(c.Request.Context(), compaction.CompactRequest{ + ChannelID: channelID, + UserID: userID, + TeamID: teamID, + Trigger: "manual", + }) if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to load conversation"}) - return - } - if len(path) < 4 { - c.JSON(http.StatusBadRequest, gin.H{"error": "conversation too short to summarize"}) - return - } - - // Find existing summary boundary (if any) and only summarize after it - startIdx := 0 - for i, m := range path { - if isSummaryMessage(&m) { - startIdx = i + 1 + // Map specific errors to HTTP status codes + switch err.Error() { + case "conversation too short to summarize": + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + case "not enough new messages since last summary": + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + default: + c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()}) } - } - - // Build messages to summarize (skip system messages, skip previous summaries) - var toSummarize []string - messagesInScope := 0 - var lastMessageID string - for _, m := range path[startIdx:] { - if m.Role == "system" || isSummaryMessage(&m) { - continue - } - toSummarize = append(toSummarize, fmt.Sprintf("%s: %s", m.Role, m.Content)) - messagesInScope++ - lastMessageID = m.ID - } - - if messagesInScope < 4 { - c.JSON(http.StatusBadRequest, gin.H{"error": "not enough new messages since last summary"}) return } - // ── Build the summarization prompt ── - conversationText := strings.Join(toSummarize, "\n\n") - summaryPrompt := []providers.Message{ - { - Role: "system", - Content: `You are a conversation summarizer. Create a concise but comprehensive summary of the following conversation that preserves: -- Key decisions and conclusions reached -- Important facts, names, numbers, and technical details mentioned -- Action items or commitments made -- The overall context and topic flow - -Write the summary in a way that would allow the conversation to continue naturally. Use clear, factual language. Do not include meta-commentary about the summarization process.`, - }, - { - Role: "user", - Content: "Summarize this conversation:\n\n" + conversationText, - }, - } - - // ── Resolve team context for the user ── - teamID := h.getUserTeamID(c.Request.Context(), userID) - - // ── Call utility role ── - log.Printf("📝 Summarizing channel %s for user %s (%d messages)", channelID, userID, messagesInScope) - result, err := h.resolver.Complete(c.Request.Context(), roles.RoleUtility, userID, teamID, summaryPrompt) - if err != nil { - log.Printf("⚠ Summarize failed for channel %s: %v", channelID, err) - c.JSON(http.StatusBadGateway, gin.H{"error": "summarization failed: " + err.Error()}) - return - } - - // ── Log usage ── - h.logSummaryUsage(c.Request.Context(), channelID, userID, result) - - // ── Insert summary message as a tree node ── - // The summary message is inserted as an "assistant" message with special metadata. - // Its parent is the last message that was summarized, making it part of the tree. - metadata := models.JSONMap{ - "type": "summary", - "summarized_until_id": lastMessageID, - "summarized_count": messagesInScope, - "utility_model": result.Model, - "used_fallback": result.UsedFallback, - } - - metaJSON, _ := json.Marshal(metadata) - siblingIdx := nextSiblingIndex(channelID, &lastMessageID) - - var summaryMsgID string - err = database.DB.QueryRow(` - INSERT INTO messages (channel_id, parent_id, role, content, model, metadata, sibling_index, participant_type) - VALUES ($1, $2, 'assistant', $3, $4, $5, $6, 'system') - RETURNING id - `, channelID, lastMessageID, result.Content, result.Model, string(metaJSON), siblingIdx).Scan(&summaryMsgID) - if err != nil { - log.Printf("⚠ Failed to persist summary for channel %s: %v", channelID, err) - c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to save summary"}) - return - } - - // Update cursor to point to the summary node - if err := updateCursor(channelID, userID, summaryMsgID); err != nil { - log.Printf("⚠ Failed to update cursor after summarize: %v", err) - } - - log.Printf("✅ Summary created for channel %s: %s (%d messages → %d chars)", - channelID, summaryMsgID, messagesInScope, len(result.Content)) - c.JSON(http.StatusOK, gin.H{ - "summary_id": summaryMsgID, - "summarized_count": messagesInScope, + "summary_id": result.SummaryID, + "summarized_count": result.SummarizedCount, "model": result.Model, "used_fallback": result.UsedFallback, "content": result.Content, }) } - -// ── Rate Limiting ───────────────────────── - -func (h *SummarizeHandler) checkRateLimit(ctx context.Context, userID string) error { - // Load rate limit from global settings (default: 20/hour, 0 = unlimited) - limit := 20 - settings, err := h.stores.GlobalConfig.Get(ctx, "utility_rate_limit") - if err == nil { - if v, ok := settings["value"]; ok { - switch n := v.(type) { - case float64: - limit = int(n) - case int: - limit = n - } - } - } - - if limit <= 0 { - return nil // unlimited - } - - since := time.Now().Add(-1 * time.Hour) - count, err := h.stores.Usage.CountRecentByRole(ctx, userID, roles.RoleUtility, since) - if err != nil { - log.Printf("⚠ Rate limit check failed: %v", err) - return nil // fail open — don't block on DB errors - } - - if count >= limit { - return fmt.Errorf("rate limit exceeded: %d utility calls in the last hour (limit: %d)", count, limit) - } - return nil -} - -// ── Helpers ─────────────────────────────── - -func (h *SummarizeHandler) getUserTeamID(ctx context.Context, userID string) *string { - // Get the user's first team (for role override resolution) - var teamID string - err := database.DB.QueryRow(` - SELECT team_id FROM team_members WHERE user_id = $1 LIMIT 1 - `, userID).Scan(&teamID) - if err != nil || teamID == "" { - return nil - } - return &teamID -} - -func (h *SummarizeHandler) logSummaryUsage(ctx context.Context, channelID, userID string, result *roles.CompletionResult) { - role := roles.RoleUtility - entry := &models.UsageEntry{ - ChannelID: &channelID, - UserID: userID, - ProviderConfigID: &result.ConfigID, - ProviderScope: result.ProviderScope, - ModelID: result.Model, - Role: &role, - PromptTokens: result.InputTokens, - CompletionTokens: result.OutputTokens, - CacheCreationTokens: result.CacheCreationTokens, - CacheReadTokens: result.CacheReadTokens, - } - - // Calculate cost from pricing - pricing, err := h.stores.Pricing.GetForModel(ctx, result.ConfigID, result.Model) - if err == nil && pricing != nil { - if pricing.InputPerM != nil { - costIn := float64(result.InputTokens) / 1_000_000 * *pricing.InputPerM - entry.CostInput = &costIn - } - if pricing.OutputPerM != nil { - costOut := float64(result.OutputTokens) / 1_000_000 * *pricing.OutputPerM - entry.CostOutput = &costOut - } - } - - if err := h.stores.Usage.Log(ctx, entry); err != nil { - log.Printf("⚠ Failed to log summary usage: %v", err) - } -} diff --git a/server/handlers/tree.go b/server/handlers/tree.go index 227fae9..5656ed1 100644 --- a/server/handlers/tree.go +++ b/server/handlers/tree.go @@ -1,309 +1,59 @@ package handlers -import ( - "database/sql" - "encoding/json" - "fmt" +// This file previously contained all tree traversal logic (path building, +// sibling queries, cursor management). As of v0.15.0, the core logic lives +// in the treepath package; these are package-local aliases so existing +// handler code (messages.go, completion.go) compiles with minimal churn. +// +// New code should import treepath directly. - "git.gobha.me/xcaliber/chat-switchboard/database" +import ( + "git.gobha.me/xcaliber/chat-switchboard/treepath" ) -// ── Tree Types ────────────────────────────── +// ── Type Aliases ──────────────────────────── +// Existing handler code references these types by unqualified name. -// PathMessage represents a single message in an active path, enriched -// with sibling metadata for the frontend branch indicator. -type PathMessage struct { - ID string `json:"id"` - ParentID *string `json:"parent_id"` - Role string `json:"role"` - Content string `json:"content"` - Model *string `json:"model"` - TokensUsed *int `json:"tokens_used,omitempty"` - ToolCalls *json.RawMessage `json:"tool_calls,omitempty"` - Metadata *json.RawMessage `json:"metadata,omitempty"` - SiblingCount int `json:"sibling_count"` - SiblingIndex int `json:"sibling_index"` - ParticipantType string `json:"participant_type,omitempty"` - ParticipantID string `json:"participant_id,omitempty"` - CreatedAt string `json:"created_at"` -} +type PathMessage = treepath.PathMessage +type SiblingInfo = treepath.SiblingInfo -// SiblingInfo is a lightweight entry for the siblings listing endpoint. -type SiblingInfo struct { - ID string `json:"id"` - Role string `json:"role"` - Model *string `json:"model,omitempty"` - SiblingIndex int `json:"sibling_index"` - Preview string `json:"preview"` // first ~80 chars of content - CreatedAt string `json:"created_at"` -} +// ── Function Wrappers ─────────────────────── -// ── Get Active Leaf ───────────────────────── - -// getActiveLeaf returns the cursor's active_leaf_id for a user in a channel. -// Falls back to the chronologically latest live message if no cursor exists. func getActiveLeaf(channelID, userID string) (*string, error) { - var leafID *string - - // Try cursor first - err := database.DB.QueryRow(` - SELECT active_leaf_id FROM channel_cursors - WHERE channel_id = $1 AND user_id = $2 - `, channelID, userID).Scan(&leafID) - - if err == nil && leafID != nil { - // Verify the leaf still exists and isn't deleted - var exists bool - database.DB.QueryRow(` - SELECT EXISTS(SELECT 1 FROM messages WHERE id = $1 AND deleted_at IS NULL) - `, *leafID).Scan(&exists) - if exists { - return leafID, nil - } - } - - // Fallback: latest live message in channel - var fallbackID string - err = database.DB.QueryRow(` - SELECT id FROM messages - WHERE channel_id = $1 AND deleted_at IS NULL - ORDER BY created_at DESC LIMIT 1 - `, channelID).Scan(&fallbackID) - if err == sql.ErrNoRows { - return nil, nil // empty channel - } - if err != nil { - return nil, err - } - return &fallbackID, nil + return treepath.GetActiveLeaf(channelID, userID) } -// ── Get Active Path ───────────────────────── - -// getActivePath walks from the active leaf up to the root via parent_id, -// returning messages in root-first order. This is what gets sent to the LLM. func getActivePath(channelID, userID string) ([]PathMessage, error) { - leafID, err := getActiveLeaf(channelID, userID) - if err != nil { - return nil, err - } - if leafID == nil { - return []PathMessage{}, nil // empty channel - } - - return getPathToLeaf(channelID, *leafID) + return treepath.GetActivePath(channelID, userID) } -// getPathToLeaf walks from a specific leaf up to root, returning root-first. func getPathToLeaf(channelID, leafID string) ([]PathMessage, error) { - rows, err := database.DB.Query(` - WITH RECURSIVE path AS ( - -- Anchor: start at the leaf - SELECT id, parent_id, role, content, model, tokens_used, tool_calls, metadata, - participant_type, participant_id, sibling_index, created_at, - 0 AS depth - FROM messages - WHERE id = $1 AND channel_id = $2 AND deleted_at IS NULL - - UNION ALL - - -- Walk up via parent_id - SELECT m.id, m.parent_id, m.role, m.content, m.model, m.tokens_used, m.tool_calls, m.metadata, - m.participant_type, m.participant_id, m.sibling_index, m.created_at, - p.depth + 1 - FROM messages m - JOIN path p ON m.id = p.parent_id - WHERE m.deleted_at IS NULL - ) - SELECT id, parent_id, role, content, model, tokens_used, tool_calls, metadata, - participant_type, participant_id, sibling_index, created_at - FROM path - ORDER BY depth DESC - `, leafID, channelID) - if err != nil { - return nil, fmt.Errorf("getPathToLeaf: %w", err) - } - defer rows.Close() - - var path []PathMessage - for rows.Next() { - var m PathMessage - var participantType, participantID sql.NullString - var toolCallsJSON, metadataJSON []byte - if err := rows.Scan( - &m.ID, &m.ParentID, &m.Role, &m.Content, &m.Model, &m.TokensUsed, - &toolCallsJSON, &metadataJSON, - &participantType, &participantID, &m.SiblingIndex, &m.CreatedAt, - ); err != nil { - return nil, fmt.Errorf("getPathToLeaf scan: %w", err) - } - if len(toolCallsJSON) > 0 && string(toolCallsJSON) != "null" { - raw := json.RawMessage(toolCallsJSON) - m.ToolCalls = &raw - } - if len(metadataJSON) > 0 && string(metadataJSON) != "null" && string(metadataJSON) != "{}" { - raw := json.RawMessage(metadataJSON) - m.Metadata = &raw - } - if participantType.Valid { - m.ParticipantType = participantType.String - } - if participantID.Valid { - m.ParticipantID = participantID.String - } - path = append(path, m) - } - - // Enrich with sibling counts - for i := range path { - path[i].SiblingCount = getSiblingCount(channelID, path[i].ParentID) - } - - return path, nil + return treepath.GetPathToLeaf(channelID, leafID) } -// ── Sibling Queries ───────────────────────── - -// getSiblingCount returns how many live siblings share the same parent_id. -// Root messages (parent_id IS NULL) count all roots in the same channel. func getSiblingCount(channelID string, parentID *string) int { - var count int - if parentID == nil { - database.DB.QueryRow(` - SELECT COUNT(*) FROM messages - WHERE channel_id = $1 AND parent_id IS NULL AND deleted_at IS NULL - `, channelID).Scan(&count) - } else { - database.DB.QueryRow(` - SELECT COUNT(*) FROM messages - WHERE parent_id = $1 AND deleted_at IS NULL - `, *parentID).Scan(&count) - } - if count == 0 { - count = 1 - } - return count + return treepath.GetSiblingCount(channelID, parentID) } -// getSiblings returns all live siblings of a given message (including itself), -// ordered by sibling_index. func getSiblings(messageID string) ([]SiblingInfo, int, error) { - // First get the parent_id of the target message - var parentID *string - var channelID string - err := database.DB.QueryRow(` - SELECT parent_id, channel_id FROM messages - WHERE id = $1 AND deleted_at IS NULL - `, messageID).Scan(&parentID, &channelID) - if err != nil { - return nil, 0, fmt.Errorf("message not found: %w", err) - } - - var rows *sql.Rows - if parentID == nil { - // Root messages: siblings are other roots in the same channel - rows, err = database.DB.Query(` - SELECT id, role, model, sibling_index, LEFT(content, 80), created_at - FROM messages - WHERE channel_id = $1 AND parent_id IS NULL AND deleted_at IS NULL - ORDER BY sibling_index, created_at - `, channelID) - } else { - rows, err = database.DB.Query(` - SELECT id, role, model, sibling_index, LEFT(content, 80), created_at - FROM messages - WHERE parent_id = $1 AND deleted_at IS NULL - ORDER BY sibling_index, created_at - `, *parentID) - } - if err != nil { - return nil, 0, err - } - defer rows.Close() - - var siblings []SiblingInfo - currentIdx := 0 - for i := 0; rows.Next(); i++ { - var s SiblingInfo - if err := rows.Scan(&s.ID, &s.Role, &s.Model, &s.SiblingIndex, &s.Preview, &s.CreatedAt); err != nil { - return nil, 0, err - } - if s.ID == messageID { - currentIdx = i - } - siblings = append(siblings, s) - } - - return siblings, currentIdx, nil + return treepath.GetSiblings(messageID) } -// ── Find Leaf ─────────────────────────────── - -// findLeafFromMessage walks down from a message to find the deepest descendant. -// Follows the first child (lowest sibling_index) at each level. -// Used when switching branches: clicking a mid-tree sibling navigates to its leaf. func findLeafFromMessage(messageID string) (string, error) { - var leafID string - err := database.DB.QueryRow(` - WITH RECURSIVE descendants AS ( - SELECT id, 0 AS depth - FROM messages - WHERE id = $1 AND deleted_at IS NULL - - UNION ALL - - SELECT child.id, d.depth + 1 - FROM messages child - JOIN descendants d ON child.parent_id = d.id - WHERE child.deleted_at IS NULL - AND child.sibling_index = ( - SELECT MIN(sibling_index) FROM messages - WHERE parent_id = d.id AND deleted_at IS NULL - ) - ) - SELECT id FROM descendants - ORDER BY depth DESC - LIMIT 1 - `, messageID).Scan(&leafID) - - if err != nil { - return messageID, nil // if anything fails, just use the message itself - } - return leafID, nil + return treepath.FindLeafFromMessage(messageID) } -// ── Next Sibling Index ────────────────────── - -// nextSiblingIndex returns the next sibling_index for a new child of parentID. -// For root messages (parentID is nil), counts existing roots in the channel. func nextSiblingIndex(channelID string, parentID *string) int { - var maxIdx sql.NullInt64 - if parentID == nil { - database.DB.QueryRow(` - SELECT MAX(sibling_index) FROM messages - WHERE channel_id = $1 AND parent_id IS NULL AND deleted_at IS NULL - `, channelID).Scan(&maxIdx) - } else { - database.DB.QueryRow(` - SELECT MAX(sibling_index) FROM messages - WHERE parent_id = $1 AND deleted_at IS NULL - `, *parentID).Scan(&maxIdx) - } - if !maxIdx.Valid { - return 0 - } - return int(maxIdx.Int64) + 1 + return treepath.NextSiblingIndex(channelID, parentID) } -// ── Update Cursor ─────────────────────────── - -// updateCursor upserts the channel_cursors row for a user. func updateCursor(channelID, userID, messageID string) error { - _, err := database.DB.Exec(` - INSERT INTO channel_cursors (channel_id, user_id, active_leaf_id) - VALUES ($1, $2, $3) - ON CONFLICT (channel_id, user_id) DO UPDATE SET - active_leaf_id = $3, updated_at = NOW() - `, channelID, userID, messageID) - return err + return treepath.UpdateCursor(channelID, userID, messageID) +} + +// isSummaryMessage checks if a PathMessage has summary metadata. +// This was previously defined in completion.go; moved here alongside the +// other tree helpers so all callers use the canonical treepath version. +func isSummaryMessage(m *PathMessage) bool { + return treepath.IsSummaryMessage(m) } diff --git a/server/treepath/cursor.go b/server/treepath/cursor.go new file mode 100644 index 0000000..fac073f --- /dev/null +++ b/server/treepath/cursor.go @@ -0,0 +1,39 @@ +package treepath + +import ( + "database/sql" + + "git.gobha.me/xcaliber/chat-switchboard/database" +) + +// UpdateCursor upserts the channel_cursors row for a user. +func UpdateCursor(channelID, userID, messageID string) error { + _, err := database.DB.Exec(` + INSERT INTO channel_cursors (channel_id, user_id, active_leaf_id) + VALUES ($1, $2, $3) + ON CONFLICT (channel_id, user_id) DO UPDATE SET + active_leaf_id = $3, updated_at = NOW() + `, channelID, userID, messageID) + return err +} + +// NextSiblingIndex returns the next sibling_index for a new child of parentID. +// For root messages (parentID is nil), counts existing roots in the channel. +func NextSiblingIndex(channelID string, parentID *string) int { + var maxIdx sql.NullInt64 + if parentID == nil { + database.DB.QueryRow(` + SELECT MAX(sibling_index) FROM messages + WHERE channel_id = $1 AND parent_id IS NULL AND deleted_at IS NULL + `, channelID).Scan(&maxIdx) + } else { + database.DB.QueryRow(` + SELECT MAX(sibling_index) FROM messages + WHERE parent_id = $1 AND deleted_at IS NULL + `, *parentID).Scan(&maxIdx) + } + if !maxIdx.Valid { + return 0 + } + return int(maxIdx.Int64) + 1 +} diff --git a/server/treepath/path.go b/server/treepath/path.go new file mode 100644 index 0000000..bd2021d --- /dev/null +++ b/server/treepath/path.go @@ -0,0 +1,163 @@ +package treepath + +import ( + "database/sql" + "encoding/json" + "fmt" + + "git.gobha.me/xcaliber/chat-switchboard/database" +) + +// ── Tree Types ────────────────────────────── + +// PathMessage represents a single message in an active path, enriched +// with sibling metadata for the frontend branch indicator. +type PathMessage struct { + ID string `json:"id"` + ParentID *string `json:"parent_id"` + Role string `json:"role"` + Content string `json:"content"` + Model *string `json:"model"` + TokensUsed *int `json:"tokens_used,omitempty"` + ToolCalls *json.RawMessage `json:"tool_calls,omitempty"` + Metadata *json.RawMessage `json:"metadata,omitempty"` + SiblingCount int `json:"sibling_count"` + SiblingIndex int `json:"sibling_index"` + ParticipantType string `json:"participant_type,omitempty"` + ParticipantID string `json:"participant_id,omitempty"` + CreatedAt string `json:"created_at"` +} + +// SiblingInfo is a lightweight entry for the siblings listing endpoint. +type SiblingInfo struct { + ID string `json:"id"` + Role string `json:"role"` + Model *string `json:"model,omitempty"` + SiblingIndex int `json:"sibling_index"` + Preview string `json:"preview"` // first ~80 chars of content + CreatedAt string `json:"created_at"` +} + +// ── Get Active Leaf ───────────────────────── + +// GetActiveLeaf returns the cursor's active_leaf_id for a user in a channel. +// Falls back to the chronologically latest live message if no cursor exists. +func GetActiveLeaf(channelID, userID string) (*string, error) { + var leafID *string + + // Try cursor first + err := database.DB.QueryRow(` + SELECT active_leaf_id FROM channel_cursors + WHERE channel_id = $1 AND user_id = $2 + `, channelID, userID).Scan(&leafID) + + if err == nil && leafID != nil { + // Verify the leaf still exists and isn't deleted + var exists bool + database.DB.QueryRow(` + SELECT EXISTS(SELECT 1 FROM messages WHERE id = $1 AND deleted_at IS NULL) + `, *leafID).Scan(&exists) + if exists { + return leafID, nil + } + } + + // Fallback: latest live message in channel + var fallbackID string + err = database.DB.QueryRow(` + SELECT id FROM messages + WHERE channel_id = $1 AND deleted_at IS NULL + ORDER BY created_at DESC LIMIT 1 + `, channelID).Scan(&fallbackID) + if err == sql.ErrNoRows { + return nil, nil // empty channel + } + if err != nil { + return nil, err + } + return &fallbackID, nil +} + +// ── Get Active Path ───────────────────────── + +// GetActivePath walks from the active leaf up to the root via parent_id, +// returning messages in root-first order. This is what gets sent to the LLM. +func GetActivePath(channelID, userID string) ([]PathMessage, error) { + leafID, err := GetActiveLeaf(channelID, userID) + if err != nil { + return nil, err + } + if leafID == nil { + return []PathMessage{}, nil // empty channel + } + + return GetPathToLeaf(channelID, *leafID) +} + +// GetPathToLeaf walks from a specific leaf up to root, returning root-first. +func GetPathToLeaf(channelID, leafID string) ([]PathMessage, error) { + rows, err := database.DB.Query(` + WITH RECURSIVE path AS ( + -- Anchor: start at the leaf + SELECT id, parent_id, role, content, model, tokens_used, tool_calls, metadata, + participant_type, participant_id, sibling_index, created_at, + 0 AS depth + FROM messages + WHERE id = $1 AND channel_id = $2 AND deleted_at IS NULL + + UNION ALL + + -- Walk up via parent_id + SELECT m.id, m.parent_id, m.role, m.content, m.model, m.tokens_used, m.tool_calls, m.metadata, + m.participant_type, m.participant_id, m.sibling_index, m.created_at, + p.depth + 1 + FROM messages m + JOIN path p ON m.id = p.parent_id + WHERE m.deleted_at IS NULL + ) + SELECT id, parent_id, role, content, model, tokens_used, tool_calls, metadata, + participant_type, participant_id, sibling_index, created_at + FROM path + ORDER BY depth DESC + `, leafID, channelID) + if err != nil { + return nil, fmt.Errorf("GetPathToLeaf: %w", err) + } + defer rows.Close() + + var path []PathMessage + for rows.Next() { + var m PathMessage + var participantType, participantID sql.NullString + var toolCallsJSON, metadataJSON []byte + if err := rows.Scan( + &m.ID, &m.ParentID, &m.Role, &m.Content, &m.Model, &m.TokensUsed, + &toolCallsJSON, &metadataJSON, + &participantType, &participantID, &m.SiblingIndex, &m.CreatedAt, + ); err != nil { + return nil, fmt.Errorf("GetPathToLeaf scan: %w", err) + } + if len(toolCallsJSON) > 0 && string(toolCallsJSON) != "null" { + raw := json.RawMessage(toolCallsJSON) + m.ToolCalls = &raw + } + if len(metadataJSON) > 0 && string(metadataJSON) != "null" && string(metadataJSON) != "{}" { + raw := json.RawMessage(metadataJSON) + m.Metadata = &raw + } + if participantType.Valid { + m.ParticipantType = participantType.String + } + if participantID.Valid { + m.ParticipantID = participantID.String + } + path = append(path, m) + } + + // Enrich with sibling counts + for i := range path { + path[i].SiblingCount = GetSiblingCount(channelID, path[i].ParentID) + } + + return path, nil +} diff --git a/server/treepath/siblings.go b/server/treepath/siblings.go new file mode 100644 index 0000000..79b2178 --- /dev/null +++ b/server/treepath/siblings.go @@ -0,0 +1,114 @@ +package treepath + +import ( + "database/sql" + "fmt" + + "git.gobha.me/xcaliber/chat-switchboard/database" +) + +// GetSiblingCount returns how many live siblings share the same parent_id. +// Root messages (parent_id IS NULL) count all roots in the same channel. +func GetSiblingCount(channelID string, parentID *string) int { + var count int + if parentID == nil { + database.DB.QueryRow(` + SELECT COUNT(*) FROM messages + WHERE channel_id = $1 AND parent_id IS NULL AND deleted_at IS NULL + `, channelID).Scan(&count) + } else { + database.DB.QueryRow(` + SELECT COUNT(*) FROM messages + WHERE parent_id = $1 AND deleted_at IS NULL + `, *parentID).Scan(&count) + } + if count == 0 { + count = 1 + } + return count +} + +// GetSiblings returns all live siblings of a given message (including itself), +// ordered by sibling_index. +func GetSiblings(messageID string) ([]SiblingInfo, int, error) { + // First get the parent_id of the target message + var parentID *string + var channelID string + err := database.DB.QueryRow(` + SELECT parent_id, channel_id FROM messages + WHERE id = $1 AND deleted_at IS NULL + `, messageID).Scan(&parentID, &channelID) + if err != nil { + return nil, 0, fmt.Errorf("message not found: %w", err) + } + + var rows *sql.Rows + if parentID == nil { + // Root messages: siblings are other roots in the same channel + rows, err = database.DB.Query(` + SELECT id, role, model, sibling_index, LEFT(content, 80), created_at + FROM messages + WHERE channel_id = $1 AND parent_id IS NULL AND deleted_at IS NULL + ORDER BY sibling_index, created_at + `, channelID) + } else { + rows, err = database.DB.Query(` + SELECT id, role, model, sibling_index, LEFT(content, 80), created_at + FROM messages + WHERE parent_id = $1 AND deleted_at IS NULL + ORDER BY sibling_index, created_at + `, *parentID) + } + if err != nil { + return nil, 0, err + } + defer rows.Close() + + var siblings []SiblingInfo + currentIdx := 0 + for i := 0; rows.Next(); i++ { + var s SiblingInfo + if err := rows.Scan(&s.ID, &s.Role, &s.Model, &s.SiblingIndex, &s.Preview, &s.CreatedAt); err != nil { + return nil, 0, err + } + if s.ID == messageID { + currentIdx = i + } + siblings = append(siblings, s) + } + + return siblings, currentIdx, nil +} + +// FindLeafFromMessage walks down from a message to find the deepest descendant. +// Follows the first child (lowest sibling_index) at each level. +// Used when switching branches: clicking a mid-tree sibling navigates to its leaf. +func FindLeafFromMessage(messageID string) (string, error) { + var leafID string + err := database.DB.QueryRow(` + WITH RECURSIVE descendants AS ( + SELECT id, 0 AS depth + FROM messages + WHERE id = $1 AND deleted_at IS NULL + + UNION ALL + + SELECT child.id, d.depth + 1 + FROM messages child + JOIN descendants d ON child.parent_id = d.id + WHERE child.deleted_at IS NULL + AND child.sibling_index = ( + SELECT MIN(sibling_index) FROM messages + WHERE parent_id = d.id AND deleted_at IS NULL + ) + ) + SELECT id FROM descendants + ORDER BY depth DESC + LIMIT 1 + `, messageID).Scan(&leafID) + + if err != nil { + return messageID, nil // if anything fails, just use the message itself + } + return leafID, nil +} diff --git a/server/treepath/summary.go b/server/treepath/summary.go new file mode 100644 index 0000000..5fd90b8 --- /dev/null +++ b/server/treepath/summary.go @@ -0,0 +1,38 @@ +package treepath + +import "encoding/json" + +// IsSummaryMessage checks if a PathMessage has summary metadata +// (metadata.type == "summary"). +func IsSummaryMessage(m *PathMessage) bool { + if m.Metadata == nil { + return false + } + meta := ParseMetadata(m) + return meta["type"] == "summary" +} + +// ParseMetadata unmarshals the JSONB metadata on a PathMessage. +// Returns an empty map on any error. +func ParseMetadata(m *PathMessage) map[string]interface{} { + if m.Metadata == nil { + return map[string]interface{}{} + } + var meta map[string]interface{} + if err := json.Unmarshal(*m.Metadata, &meta); err != nil { + return map[string]interface{}{} + } + return meta +} + +// FindSummaryBoundary returns the index of the most recent summary node +// in a path, or -1 if no summary exists. +func FindSummaryBoundary(path []PathMessage) int { + idx := -1 + for i := range path { + if IsSummaryMessage(&path[i]) { + idx = i + } + } + return idx +} diff --git a/src/css/styles.css b/src/css/styles.css index 71e3d5f..3e05587 100644 --- a/src/css/styles.css +++ b/src/css/styles.css @@ -862,6 +862,12 @@ a:hover { text-decoration: underline; } .context-warning-action:hover { background: var(--accent); color: var(--text-on-color); border-color: var(--accent); } .context-warning-action:disabled { opacity: 0.5; cursor: not-allowed; } .context-warning-action:disabled:hover { background: var(--bg-surface); color: var(--text); border-color: var(--border); } +.context-warning-toggle { + display: inline-flex; align-items: center; gap: 4px; flex-shrink: 0; + font-size: 0.72rem; color: var(--text-3); cursor: pointer; white-space: nowrap; + user-select: none; +} +.context-warning-toggle input[type="checkbox"] { margin: 0; cursor: pointer; } /* Summary message node */ .message-summary { border: 1px dashed color-mix(in srgb, var(--accent) 40%, transparent); diff --git a/src/index.html b/src/index.html index 35d63c2..5929f25 100644 --- a/src/index.html +++ b/src/index.html @@ -145,6 +145,9 @@ ⚠️ + @@ -769,6 +772,29 @@

Web search and URL fetch tools allow AI to look up current information. DuckDuckGo works out of the box. SearXNG is ideal for airgapped deployments.

+
+

Auto-Compaction

+ +

When enabled, long conversations are automatically summarized in the background using the utility model. Requires a utility model role to be configured.

+ +

🔐 Encryption

Loading…
diff --git a/src/js/chat.js b/src/js/chat.js index e4c765e..a66f4f0 100644 --- a/src/js/chat.js +++ b/src/js/chat.js @@ -52,6 +52,26 @@ async function summarizeAndContinue() { } } +// ── Per-Channel Auto-Compaction Toggle ────── + +async function toggleChannelAutoCompact(enabled) { + const chatId = App.currentChatId; + if (!chatId) return; + const chat = App.chats.find(c => c.id === chatId); + if (!chat) return; + + // Update local state + if (!chat.settings) chat.settings = {}; + chat.settings.auto_compaction = enabled; + + // Persist to server (JSONB merge) + try { + await API.updateChannel(chatId, { settings: { auto_compaction: enabled } }); + } catch (e) { + console.warn('Failed to save auto-compaction setting:', e.message); + } +} + // ── Chat Management ────────────────────────── async function loadChats() { diff --git a/src/js/settings-handlers.js b/src/js/settings-handlers.js index e0e5154..85d430d 100644 --- a/src/js/settings-handlers.js +++ b/src/js/settings-handlers.js @@ -58,13 +58,36 @@ function updateAvatarPreview(dataURI) { // ── Settings Handlers ──────────────────────── -function handleSaveSettings() { +async function handleSaveSettings() { App.settings.model = document.getElementById('settingsModel').value || App.settings.model; App.settings.systemPrompt = document.getElementById('settingsSystemPrompt').value.trim(); App.settings.maxTokens = parseInt(document.getElementById('settingsMaxTokens').value) || 0; // 0 = auto App.settings.temperature = parseFloat(document.getElementById('settingsTemp').value) || 0.7; App.settings.showThinking = document.getElementById('settingsThinking').checked; saveSettings(); + + // ── Save profile fields (display name, email) ── + const displayName = document.getElementById('profileDisplayName').value.trim(); + const email = document.getElementById('profileEmail').value.trim(); + const profileUpdates = {}; + if (displayName !== (API.user?.display_name || '')) profileUpdates.display_name = displayName; + if (email !== (API.user?.email || '')) profileUpdates.email = email; + if (Object.keys(profileUpdates).length > 0) { + try { + const updated = await API.updateProfile(profileUpdates); + // Sync local user object so sidebar/avatar reflect the change + if (updated) { + if (updated.display_name !== undefined) API.user.display_name = updated.display_name; + if (updated.email !== undefined) API.user.email = updated.email; + API.saveTokens(); + UI.updateUser(); + } + } catch (e) { + console.warn('Profile save failed:', e.message); + UI.toast('Profile update failed: ' + e.message, 'error'); + } + } + UI.updateModelSelector(); UI.updateCapabilityBadges(); UI.closeSettings(); @@ -204,6 +227,20 @@ async function handleSaveAdminSettings() { }; await API.adminUpdateSetting('search_config', { value: searchConfig }); + // Auto-Compaction config → global_settings + const compactionEnabled = document.getElementById('adminCompactionEnabled')?.checked || false; + const compactionThreshold = parseInt(document.getElementById('adminCompactionThreshold')?.value) || 70; + const compactionCooldown = parseInt(document.getElementById('adminCompactionCooldown')?.value) || 30; + await API.adminUpdateSetting('auto_compaction', { value: { + enabled: compactionEnabled, + threshold: compactionThreshold, + cooldown: compactionCooldown, + }}); + // Scanner reads these individual keys each tick + await API.adminUpdateSetting('auto_compaction_enabled', { value: compactionEnabled }); + await API.adminUpdateSetting('auto_compaction_threshold', { value: compactionThreshold / 100 }); + await API.adminUpdateSetting('auto_compaction_cooldown_minutes', { value: compactionCooldown }); + UI.toast('Settings saved', 'success'); // Live-apply: refresh policies and dependent UI await initBanners(); @@ -233,6 +270,9 @@ function _initUserRolePrimitive() { return list.filter(c => c.scope === 'personal'); }, fetchModels: async () => { + // Refresh App.models to ensure personal provider models are current. + // Without this, newly added providers may not appear in the dropdown. + await fetchModels(); return App.models || []; }, fetchRoles: async () => { @@ -591,6 +631,11 @@ function _initSettingsListeners() { document.getElementById('adminBannerBgHex')?.addEventListener('input', (e) => { if (/^#[0-9a-f]{6}$/i.test(e.target.value)) { document.getElementById('adminBannerBg').value = e.target.value; UI.updateBannerPreview(); }}); document.getElementById('adminBannerFgHex')?.addEventListener('input', (e) => { if (/^#[0-9a-f]{6}$/i.test(e.target.value)) { document.getElementById('adminBannerFg').value = e.target.value; UI.updateBannerPreview(); }}); + // Admin — auto-compaction controls + document.getElementById('adminCompactionEnabled')?.addEventListener('change', (e) => { + document.getElementById('compactionConfigFields').style.display = e.target.checked ? '' : 'none'; + }); + // Admin — search provider controls document.getElementById('adminSearchProvider')?.addEventListener('change', (e) => { document.getElementById('searxngConfigFields').style.display = e.target.value === 'searxng' ? '' : 'none'; diff --git a/src/js/tokens.js b/src/js/tokens.js index 4704bd8..a0805f4 100644 --- a/src/js/tokens.js +++ b/src/js/tokens.js @@ -103,6 +103,16 @@ function updateContextWarning() { summarizeBtn.style.display = showSummarize ? 'inline-block' : 'none'; } + // Show auto-compact toggle next to summarize button + const autoToggle = document.getElementById('autoCompactToggle'); + const autoCheck = document.getElementById('autoCompactCheck'); + if (autoToggle && autoCheck) { + autoToggle.style.display = showSummarize ? 'inline-flex' : 'none'; + // Reflect current channel setting + const chatSettings = chat?.settings || {}; + autoCheck.checked = chatSettings.auto_compaction !== false; // default: on + } + if (pct >= 0.9 && !Tokens._warningDismissed) { warning.style.display = 'flex'; warning.className = 'context-warning danger'; diff --git a/src/js/ui-admin.js b/src/js/ui-admin.js index 3781140..3f9e123 100644 --- a/src/js/ui-admin.js +++ b/src/js/ui-admin.js @@ -803,6 +803,19 @@ Object.assign(UI, { document.getElementById('searxngConfigFields').style.display = (searchCfg.provider === 'searxng') ? '' : 'none'; + // Auto-Compaction (global_settings) + const compactionCfg = getSetting('auto_compaction', {}) || {}; + const compactionEnabled = document.getElementById('adminCompactionEnabled'); + if (compactionEnabled) { + compactionEnabled.checked = !!compactionCfg.enabled; + document.getElementById('compactionConfigFields').style.display = + compactionCfg.enabled ? '' : 'none'; + } + const compThreshold = document.getElementById('adminCompactionThreshold'); + if (compThreshold) compThreshold.value = String(compactionCfg.threshold || 70); + const compCooldown = document.getElementById('adminCompactionCooldown'); + if (compCooldown) compCooldown.value = String(compactionCfg.cooldown || 30); + if (vaultEl) { try { const vault = await API.adminGetVaultStatus(); diff --git a/src/js/ui-core.js b/src/js/ui-core.js index f023ac9..b72aff6 100644 --- a/src/js/ui-core.js +++ b/src/js/ui-core.js @@ -361,10 +361,11 @@ const UI = { const meta = msg.metadata || {}; const count = meta.summarized_count || '?'; const model = meta.utility_model || 'utility model'; + const trigger = meta.trigger === 'auto' ? ' · auto' : ''; return `
- 📝 Conversation summary (${count} messages, by ${esc(model)}) + 📝 Conversation summary (${count} messages, by ${esc(model)}${trigger})
${formatMessage(msg.content)}
`;