switchboard/core
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Releases 9 Wiki Activity

Changeset 0.15.0 (#71)

Browse Source
This commit is contained in:
xcaliber
2026-02-26 21:19:55 +00:00
parent e2149e249d
commit 2d79ff593b
35 changed files with 3218 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files

312
server/compaction/compaction.go Normal file
View File

@@ -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)
}
}

401
server/compaction/compaction_test.go Normal file
View File

@@ -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)
}
}

57
server/compaction/estimator.go Normal file
View File

@@ -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
}

92
server/compaction/estimator_test.go Normal file
View File

@@ -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)
}
}

399
server/compaction/scanner.go Normal file
View File

@@ -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
}

15
server/compaction/testmain_test.go Normal file
View File

@@ -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)
}

62
server/database/seed_helpers.go Normal file
View File

@@ -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)
}
}

54
server/handlers/admin.go
View File

@@ -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 {

82
server/handlers/auth.go
View File

@@ -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
}

257
server/handlers/live_compaction_test.go Normal file
View File

@@ -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)
}
}

237
server/handlers/summarize.go
View File

@@ -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)
}
}

306
server/handlers/tree.go
View File

@@ -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)
}

39
server/treepath/cursor.go Normal file
View File

@@ -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
}

163
server/treepath/path.go Normal file
View File

@@ -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
}

114
server/treepath/siblings.go Normal file
View File

@@ -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
}

38
server/treepath/summary.go Normal file
View File

@@ -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
}