diff --git a/internal/channels/history.go b/internal/channels/history.go index fb077ac6..4360945f 100644 --- a/internal/channels/history.go +++ b/internal/channels/history.go @@ -27,8 +27,9 @@ const maxHistoryKeys = 1000 const DefaultGroupHistoryLimit = 100 const ( - flushInterval = 3 * time.Second // periodic flush interval - flushBatchMax = 20 // flush when buffer reaches this size + flushInterval = 3 * time.Second // periodic flush interval + flushBatchMax = 20 // flush when buffer reaches this size + compactSweepInterval = 10 * time.Minute // periodic compaction sweep for post-restart safety ) // HistoryEntry represents a single tracked group message. diff --git a/internal/channels/history_compaction.go b/internal/channels/history_compaction.go index 5306d499..04388635 100644 --- a/internal/channels/history_compaction.go +++ b/internal/channels/history_compaction.go @@ -129,6 +129,37 @@ func CompactGroup(ctx context.Context, s store.PendingMessageStore, channelName, return remaining, nil } +// sweepCompaction checks DB for groups exceeding compaction threshold. +// Called periodically from flushLoop as a safety net for post-restart scenarios +// where RAM is empty but DB has accumulated messages. +func (ph *PendingHistory) sweepCompaction() { + cfg := ph.compactionCfg + if cfg == nil || cfg.Provider == nil || ph.store == nil { + return + } + threshold := cfg.Threshold + if threshold <= 0 { + threshold = DefaultGroupHistoryLimit + } + + ctx, cancel := context.WithTimeout(ph.tenantCtx(), 30*time.Second) + defer cancel() + + groups, err := ph.store.ListGroups(ctx) + if err != nil { + slog.Warn("compaction.sweep_failed", "channel", ph.channelName, "error", err) + return + } + for _, g := range groups { + if g.ChannelName != ph.channelName { + continue + } + if g.MessageCount > threshold { + ph.MaybeCompact(g.HistoryKey, g.MessageCount, cfg) + } + } +} + // runCompaction performs LLM-based summarization of old messages. // Wraps CompactGroup with flush + RAM update + threshold check. func (ph *PendingHistory) runCompaction(historyKey string, cfg *CompactionConfig) { diff --git a/internal/channels/history_flush.go b/internal/channels/history_flush.go index b19a5cf5..eb424c47 100644 --- a/internal/channels/history_flush.go +++ b/internal/channels/history_flush.go @@ -73,11 +73,15 @@ func (ph *PendingHistory) signalFlush() { } // flushLoop runs in background, flushing buffer periodically or when signaled. +// Also runs periodic compaction sweep as a safety net for post-restart scenarios. func (ph *PendingHistory) flushLoop() { defer close(ph.stopped) ticker := time.NewTicker(flushInterval) defer ticker.Stop() + compactTicker := time.NewTicker(compactSweepInterval) + defer compactTicker.Stop() + for { select { case <-ph.stopCh: @@ -87,6 +91,8 @@ func (ph *PendingHistory) flushLoop() { ph.flushNow() case <-ticker.C: ph.flushNow() + case <-compactTicker.C: + ph.sweepCompaction() } } } diff --git a/internal/channels/instance_loader.go b/internal/channels/instance_loader.go index 07a00d2c..9022d9db 100644 --- a/internal/channels/instance_loader.go +++ b/internal/channels/instance_loader.go @@ -282,6 +282,16 @@ func (l *InstanceLoader) loadInstance(ctx context.Context, inst store.ChannelIns pc.SetPendingCompaction(cc) slog.Debug("pending compaction configured", "channel", inst.Name, "provider", p.Name(), "model", model, "threshold", cc.Threshold, "keep_recent", cc.KeepRecent, "max_tokens", cc.MaxTokens) + } else { + attemptedProvider := "" + if l.pendingCompactCfg != nil { + attemptedProvider = l.pendingCompactCfg.Provider + } + if attemptedProvider == "" && ag != nil { + attemptedProvider = ag.Provider + } + slog.Warn("pending compaction not configured: provider/model unavailable", + "channel", inst.Name, "agent_id", inst.AgentID, "attempted_provider", attemptedProvider) } } l.manager.RegisterChannel(inst.Name, ch)