From 4cf66eb3791fa15f92235921ae1c67fbddc33e61 Mon Sep 17 00:00:00 2001 From: viettranx Date: Fri, 10 Apr 2026 13:32:02 +0700 Subject: [PATCH] feat(ts-port): reasoning strip, dreaming config + weighted scoring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 6 — Reasoning token stripping: - ReasoningDecision.StripThinking auto-flags Kimi + DeepSeek-Reasoner - Guard clauses in Anthropic/OpenAI/Codex stream handlers - Usage.ThinkingTokens + RawAssistantContent preserved (billing + tool passback safe) Phase 8 — Per-agent dreaming config: - MemoryConfig.Dreaming JSONB (no migration), resolver callback pattern - Enabled/DebounceMs/Threshold/VerboseLog fields with partial-override merge - ConsolidationDeps gains optional AgentStore Phase 10 — Dreaming weighted scoring: - Migration 000045 adds recall_count/recall_score/last_recalled_at on episodic_summaries - ComputeRecallScore 4-component formula (freq/rel/recency/freshness, 14d half-life) - memory_search fire-and-forget RecordRecall; ListUnpromotedScored in DreamingWorker - Bootstrap-friendly filter: unrecalled entries bypass thresholds - Debounce stamped on filter-empty skip to prevent starvation loop Phase 5 follow-up — last_compaction_at in sessions.metadata JSONB: - v3 PruneStage.CompactMessages and v2 maybeSummarize both stamp timestamp - Zero migration; exported const SessionMetaKeyLastCompactionAt RequiredSchemaVersion: 44 → 45 (PG), SchemaVersion: 12 → 13 (SQLite). 27 new tests; builds pass under PG and sqliteonly tags. --- docs/06-store-data-model.md | 34 ++- docs/12-extended-thinking.md | 38 +++- docs/17-changelog.md | 53 +++++ internal/agent/loop_history_sanitize.go | 5 + internal/agent/loop_pipeline_callbacks.go | 21 +- internal/config/config.go | 16 ++ internal/consolidation/dreaming_config.go | 84 ++++++++ .../consolidation/dreaming_config_test.go | 197 ++++++++++++++++++ .../consolidation/dreaming_scoring_test.go | 191 +++++++++++++++++ internal/consolidation/dreaming_worker.go | 100 +++++++-- internal/consolidation/scoring.go | 128 ++++++++++++ internal/consolidation/scoring_test.go | 160 ++++++++++++++ internal/consolidation/workers.go | 5 + internal/consolidation/workers_test.go | 43 +++- internal/providers/anthropic.go | 17 +- internal/providers/anthropic_stream.go | 15 +- internal/providers/codex.go | 22 +- internal/providers/openai_chat.go | 16 +- internal/providers/reasoning_resolution.go | 27 ++- .../providers/reasoning_strip_stream_test.go | 102 +++++++++ internal/providers/reasoning_strip_test.go | 67 ++++++ internal/providers/types.go | 4 + internal/store/episodic_store.go | 42 ++-- internal/store/pg/episodic_search.go | 20 +- internal/store/pg/episodic_summaries.go | 62 +++++- internal/store/pg/memory_scan_rows.go | 54 ++--- internal/store/sqlitestore/episodic.go | 78 ++++++- internal/store/sqlitestore/schema.go | 10 +- internal/store/sqlitestore/schema.sql | 9 +- internal/tools/memory.go | 37 ++++ internal/upgrade/version.go | 2 +- .../000045_episodic_recall_tracking.down.sql | 4 + .../000045_episodic_recall_tracking.up.sql | 15 ++ 33 files changed, 1555 insertions(+), 123 deletions(-) create mode 100644 internal/consolidation/dreaming_config.go create mode 100644 internal/consolidation/dreaming_config_test.go create mode 100644 internal/consolidation/dreaming_scoring_test.go create mode 100644 internal/consolidation/scoring.go create mode 100644 internal/consolidation/scoring_test.go create mode 100644 internal/providers/reasoning_strip_stream_test.go create mode 100644 internal/providers/reasoning_strip_test.go create mode 100644 migrations/000045_episodic_recall_tracking.down.sql create mode 100644 migrations/000045_episodic_recall_tracking.up.sql diff --git a/docs/06-store-data-model.md b/docs/06-store-data-model.md index 1c69fb4b..e2d03547 100644 --- a/docs/06-store-data-model.md +++ b/docs/06-store-data-model.md @@ -101,6 +101,18 @@ flowchart TD | Cron | `agent:{agentId}:cron:{jobId}:run:{runId}` | `agent:default:cron:reminder:run:abc123` | | Main | `agent:{agentId}:{mainKey}` | `agent:default:main` | +### Session Metadata - Compaction Tracking + +**New well-known metadata key** (Phase 5 follow-up): `last_compaction_at` (RFC3339 string) + +This timestamp is written to `sessions.metadata` JSONB after successful message compaction (context pruning). Both execution paths update it: +- **V3 pipeline**: `PruneStage.CompactMessages()` after successful compaction +- **V2 legacy**: `maybeSummarize()` goroutine after successful summarization + +Operators can read this via `GetSessionMetadata()` to understand when a session was last compacted. The web UI optionally displays this timestamp in a context-usage tooltip. + +Go constant export: `agent.SessionMetaKeyLastCompactionAt = "last_compaction_at"` + --- ## 4. Agent Access Control @@ -661,7 +673,7 @@ L0 (Working Memory) L1 (Episodic Memory) L2 (Semantic Memory) | Table | Purpose | Key Columns | |-------|---------|-------------| -| `episodic_summaries` | Session conversation summaries | `agent_id`, `user_id`, `session_key`, `summary`, `l0_abstract`, `key_topics` (TEXT[]), `embedding` (vector), `source_id` (dedup), `expires_at` | +| `episodic_summaries` | Session conversation summaries | `agent_id`, `user_id`, `session_key`, `summary`, `l0_abstract`, `key_topics` (TEXT[]), `embedding` (vector), `source_id` (dedup), `expires_at`, `recall_count` (INT), `recall_score` (FLOAT), `last_recalled_at` (TIMESTAMPTZ) | | `agent_evolution_metrics` | Self-evolution performance data | `agent_id`, `session_key`, `metric_type` (retrieval/tool/feedback), `metric_key`, `value` (JSONB) | | `agent_evolution_suggestions` | Data-driven improvement suggestions | `agent_id`, `suggestion_type`, `suggestion`, `rationale`, `parameters` (JSONB), `status` (pending/approved/rejected/applied) | | `vault_documents` | Knowledge Vault document registry | `agent_id`, `scope` (personal/team/shared), `path`, `title`, `doc_type`, `content_hash`, `embedding` (vector), `metadata` (JSONB) | @@ -757,7 +769,25 @@ flowchart TD | **EpisodicWorker** | `run.completed` | Extract session summary via LLM or compaction summary. Generate L0 abstract. Store in `episodic_summaries`. Emit `episodic.created` | | **SemanticWorker** | `episodic.created` | Parse summary for entity mentions and relationships. Extract via regex/NER. Insert into KG tables (`kg_entities`, `kg_relations`). Emit `entity.upserted` | | **DedupWorker** | `entity.upserted` | Check for duplicate entities via embedding similarity. Merge duplicate nodes by redirecting relations. Update timestamps to reflect consolidation | -| **DreamingWorker** | `episodic.created` (debounced 10m) | Batch collect unpromoted episodic summaries. Call LLM for synthesis/insight pass. Write results to long-term memory (update KG, write to vault, etc.) | +| **DreamingWorker** | `episodic.created` (debounced 10m) | Batch collect unpromoted episodic summaries scored by usefulness (recall signal). Call LLM for synthesis/insight pass. Write results to long-term memory (update KG, write to vault, etc.) | + +### Dreaming Weighted Scoring (Phase 10, Migration 000045) + +The DreamingWorker prioritizes unpromoted episodic summaries by usefulness via a 4-component running-average score: + +**ComputeRecallScore formula** (14-day half-life): +``` +score = 0.30 * frequency + 0.35 * relevance + 0.20 * recency + 0.15 * freshness +``` + +**Tracking columns** (added to `episodic_summaries`): +- `recall_count INT DEFAULT 0` — Number of times this summary was returned in memory searches +- `recall_score DOUBLE PRECISION DEFAULT 0` — Weighted average score (0 to 1) +- `last_recalled_at TIMESTAMPTZ` — Timestamp of most recent search hit + +**Index for DreamingWorker**: `idx_episodic_recall_unpromoted` on `(agent_id, user_id, recall_score DESC) WHERE promoted_at IS NULL`. Enables efficient `ListUnpromotedScored()` queries to fetch highest-scoring summaries first. + +**Integration with memory_search tool**: After search results are returned to agent, a fire-and-forget task increments `recall_count`, updates `recall_score` via running average, and sets `last_recalled_at`. No blocking — search returns immediately. ### Configuration diff --git a/docs/12-extended-thinking.md b/docs/12-extended-thinking.md index bd2808b9..687d90de 100644 --- a/docs/12-extended-thinking.md +++ b/docs/12-extended-thinking.md @@ -227,7 +227,43 @@ OpenAI-compatible providers handle thinking/reasoning content as metadata. The ` --- -## 6. Observability +## 6. Reasoning Content Stripping (Phase 6 — OpenClaw TS port) + +Some models emit chain-of-thought reasoning tokens even when `effort="off"` is specified. To prevent that raw CoT from reaching end users, GoClaw supports a `StripThinking` flag on `ReasoningDecision`. + +### Models Known to Leak CoT + +Auto-flagged via `modelLeaksReasoning(model)` in `internal/providers/reasoning_resolution.go`: +- **Kimi family**: any model name containing `kimi` (case-insensitive, e.g. `kimi-k2`, `moonshot/kimi-k2-thinking`) +- **DeepSeek-Reasoner**: any model name containing `deepseek-reasoner` + +The allowlist is a simple substring check — extendable as new leaky models appear. + +### Auto-Enable Flow + +1. `ResolveReasoningDecision` runs its normal flow, then a `defer` checks: if `EffectiveEffort == "off" && modelLeaksReasoning(model)`, it sets `decision.StripThinking = true`. +2. The agent loop (`loop_pipeline_callbacks.go`) reads `decision.StripThinking` and propagates it into `chatReq.Options[providers.OptStripThinking] = true`. +3. Each provider's streaming handler reads `OptStripThinking` from options at the top of `ChatStream`/`Chat` and applies guard clauses. + +### Implementation Per Provider + +- **Anthropic** (`anthropic_stream.go`, `anthropic.go`): `thinking_delta` events skip `result.Thinking` accumulation AND the `onChunk` emit when stripping, but `thinkingChars` still increments so `Usage.ThinkingTokens` stays billable. `RawAssistantContent` (content blocks for tool-use passback) is never touched. Non-streaming `Chat()` clears `resp.Thinking` post-parse. +- **OpenAI** (`openai_chat.go`): streaming guards `reasoning`/`reasoning_content` delta accumulation; non-streaming clears `resp.Thinking` post-parse. `Usage.ThinkingTokens` read from the usage chunk independently. +- **Codex** (`codex.go`): `processSSEEvent` takes an extra `stripThinking bool` param; the `reasoning` item case skips summary text appending when set. Usage still extracted from `response.completed`. +- **DashScope**: inherits both guards via `OpenAIProvider` embedding — no separate implementation needed. + +### Invariants Preserved + +| Field | Stripping effect | Rationale | +|---|---|---| +| `ChatResponse.Thinking` | Cleared | User-visible output | +| `Usage.ThinkingTokens` | **Unchanged** | Billing accuracy (Phase 1 depends on it) | +| `RawAssistantContent` | **Unchanged** | Anthropic tool-use replay requires raw thinking blocks | +| `onChunk(StreamChunk{Thinking})` | Not emitted | Streaming UI display | + +--- + +## 7. Observability Each LLM span can now include a `metadata.reasoning` section with: - `source` diff --git a/docs/17-changelog.md b/docs/17-changelog.md index af8b1f09..53d1e186 100644 --- a/docs/17-changelog.md +++ b/docs/17-changelog.md @@ -32,6 +32,59 @@ All notable changes to GoClaw Gateway are documented here. Format follows [Keep ## [Unreleased] +### Added + +#### Episodic Memory Weighted Scoring — Dreaming Enhancement (2026-04-10, Phase 10) +- **Recall signal tracking**: `episodic_summaries` table gains 3 columns: `recall_count INT`, `recall_score DOUBLE PRECISION`, `last_recalled_at TIMESTAMPTZ` to track usefulness of each memory +- **ComputeRecallScore formula**: 4-component running average (30% frequency + 35% relevance + 20% recency + 15% freshness, 14-day half-life) quantifies memory value +- **DreamingWorker prioritization**: `ListUnpromotedScored()` queries sort by `recall_score DESC` instead of `created_at ASC`, promoting high-signal summaries for synthesis +- **fire-and-forget updates**: `memory_search` tool fire-and-forget tasks increment recall counts asynchronously without blocking search results +- **Index optimization**: New partial index `idx_episodic_recall_unpromoted ON episodic_summaries(agent_id, user_id, recall_score DESC) WHERE promoted_at IS NULL` for efficient DreamingWorker queries +- **Migration 000045**: PG schema v44→45 + SQLite schema v12→13 + +#### Compaction Telemetry — Message Context Tracking (2026-04-10, Phase 5 Follow-up) +- **Session metadata tracking**: `sessions.metadata` JSONB gains well-known key `last_compaction_at` (RFC3339 timestamp) after successful message compaction +- **Dual execution paths**: Both v3 `PruneStage.CompactMessages` and v2 legacy `maybeSummarize` update timestamp on successful compaction +- **Operator visibility**: `GetSessionMetadata()` exposes compaction timestamp; web UI shows in context-usage tooltip +- **Go constant export**: `agent.SessionMetaKeyLastCompactionAt = "last_compaction_at"` + +#### Provider Reasoning Content Stripping (2026-04-10, Phase 6) +- **Auto-strip known leakers**: Models known to leak chain-of-thought at effort="off" (Kimi family, DeepSeek-Reasoner) auto-enable `StripThinking` so user-visible `ChatResponse.Thinking` stays empty +- **Multi-provider support**: Guard clauses in Anthropic streaming, Anthropic non-streaming `Chat()`, OpenAI `ChatStream`/`Chat`, Codex `processSSEEvent`; DashScope inherits via OpenAIProvider embedding +- **Billing-safe invariants**: `Usage.ThinkingTokens` still counted from raw bytes; `RawAssistantContent` untouched so Anthropic tool-use passback continues to work +- **Option plumbing**: New `providers.OptStripThinking` key propagated via `ChatRequest.Options`; `ReasoningDecision.StripThinking` auto-set in `ResolveReasoningDecision` defer +- **Helper**: `modelLeaksReasoning(model string) bool` — extensible allowlist + +#### Dreaming Config Per-Agent (2026-04-10, Phase 8) +- **Per-agent overrides**: `MemoryConfig.Dreaming` JSONB on `agents.memory_config` (nested, no migration) controls dreaming worker behaviour per-agent +- **Fields**: `Enabled *bool`, `DebounceMs int`, `Threshold int`, `VerboseLog *bool` — all pointer/zero-default for partial-override merge semantics +- **Resolver pattern**: `DreamingConfigResolver func(ctx, agentID) *DreamingConfig` wired via `newAgentStoreResolver(AgentStore)`. `ConsolidationDeps` gains optional `AgentStore store.AgentCRUDStore` +- **Backward compat**: Nil resolver → struct defaults; empty JSONB → defaults via zero-value short-circuit +- **Merge helper**: `mergeDreamingConfig` applies override fields only when explicitly set, preserving base values + +#### Per-Provider Context Window & Hardening (2026-04-10, Phase 4, commit 8d37dc45) +- **EffectiveContextWindow**: Resolved once per run in ContextStage from `ModelRegistry` (via provider+model lookup); `PruneStage` uses it with fallback to `Config.ContextWindow` +- **ReserveTokens safety buffer**: New `PipelineConfig.ReserveTokens` subtracted from history budget so PruneStage compacts slightly before the hard limit +- **InMemoryCache hardening**: TTL sweep (60s) + max-size cap (10k) with oldest-first eviction; `Close()` wired to gateway shutdown +- **ContactCollector tenant fix**: Cache key now includes `tenantID + channelInstance` (was silently skipping upserts for same sender across tenants) + +#### Context-Aware Auto-Inject Query (2026-04-10, Phase 9, commit 2731f99a) +- **RecentContext enrichment**: `InjectParams.RecentContext` field supplies last 1-2 user turns; `pgAutoInjector.Inject` prepends "Context:" frame before "Query:" focus +- **Helper**: `pipeline.buildRecentContext()` walks history backward, picks last 2 user turns, caps at 300 chars +- **Callback signature**: `AutoInject func(ctx, msg, userID, recentContext string)` — additive, backward compatible +- **Why**: Vector search on "what's my favorite?" alone returns nothing useful; context-aware query captures conversational intent + +#### Cost Calculation Thinking Tokens Fix (2026-04-10, Phase 1, commit 77a80680) +- **Critical billing bug**: `CalculateCost` now properly handles `ThinkingTokens` as sub-count of `CompletionTokens` (not double-counted for OpenAI o3/o4-mini, Codex/GPT-5; properly split for Anthropic extended thinking) +- **Provider-aware**: Splits only when `ReasoningPerMillion > 0`, otherwise leaves as-is (default matches provider billing) + +#### Web UI Enhancements (2026-04-10, Phase 11) +- **Context usage badge**: Chat top bar shows `{used}/{max} ({percent}%)` with color ramp (amber ≥75%, destructive ≥90%); hidden on mobile via `hidden sm:flex` +- **Compaction indicator**: Context badge tooltip includes compaction count + last compaction timestamp (read from `session.metadata.last_compaction_at`) +- **DreamingConfig UI**: Agent detail MemorySection renders nested dreaming block (4 controls: enabled, threshold, debounce_ms, verbose_log) +- **i18n**: New keys in en/vi/zh: `agents.configSections.dreaming.*`, `chat.contextUsage.*` +- **Not shipped (YAGNI)**: "Memory recall config section" (existing MemorySection already covers all MemoryConfig fields), "Session types extension" (fields already present or out of scope) + ### Refactored #### V3 Architecture Refactor — Phase 6 Completion (2026-04-08) diff --git a/internal/agent/loop_history_sanitize.go b/internal/agent/loop_history_sanitize.go index 4830f7fe..13acea95 100644 --- a/internal/agent/loop_history_sanitize.go +++ b/internal/agent/loop_history_sanitize.go @@ -295,6 +295,11 @@ func (l *Loop) maybeSummarize(ctx context.Context, sessionKey string) { l.sessions.SetSummary(sctx, sessionKey, SanitizeAssistantContent(resp.Content)) l.sessions.TruncateHistory(sctx, sessionKey, keepLast) l.sessions.IncrementCompaction(sctx, sessionKey) + // Mirror SessionMetaKeyLastCompactionAt from the v3 prune/compact path + // so the legacy v2 post-turn summarizer also surfaces compaction cadence. + l.sessions.SetSessionMetadata(sctx, sessionKey, map[string]string{ + SessionMetaKeyLastCompactionAt: time.Now().UTC().Format(time.RFC3339), + }) l.sessions.Save(sctx, sessionKey) }() } diff --git a/internal/agent/loop_pipeline_callbacks.go b/internal/agent/loop_pipeline_callbacks.go index d3230e9c..abc93157 100644 --- a/internal/agent/loop_pipeline_callbacks.go +++ b/internal/agent/loop_pipeline_callbacks.go @@ -47,7 +47,7 @@ func (l *Loop) pipelineCallbacks(req *RunRequest, bridgeRS *runState) pipelineCa buildFilteredTools: l.makeBuildFilteredTools(req), callLLM: l.makeCallLLM(req, emitRun), pruneMessages: l.makePruneMessages(), - compactMessages: l.makeCompactMessages(), + compactMessages: l.makeCompactMessages(req), runMemoryFlush: l.makeRunMemoryFlush(), executeToolCall: l.makeExecuteToolCall(req, bridgeRS), executeToolRaw: l.makeExecuteToolRaw(req), @@ -241,6 +241,9 @@ func (l *Loop) makeCallLLM(req *RunRequest, emitRun func(AgentEvent)) func(ctx c if effort := reasoningDecision.RequestEffort(); effort != "" { chatReq.Options[providers.OptThinkingLevel] = effort } + if reasoningDecision.StripThinking { + chatReq.Options[providers.OptStripThinking] = true + } // Emit LLM span start for tracing. start := time.Now().UTC() @@ -309,16 +312,30 @@ func (l *Loop) makePruneMessages() func(msgs []providers.Message, budget int) [] } } -func (l *Loop) makeCompactMessages() func(ctx context.Context, msgs []providers.Message, model string) ([]providers.Message, error) { +func (l *Loop) makeCompactMessages(req *RunRequest) func(ctx context.Context, msgs []providers.Message, model string) ([]providers.Message, error) { return func(ctx context.Context, msgs []providers.Message, model string) ([]providers.Message, error) { compacted := l.compactMessagesInPlace(ctx, msgs) if compacted == nil { return msgs, nil // compaction failed, return original } + // Stamp session metadata with the compaction timestamp so operators + // can diagnose compaction cadence without a dedicated column. Stored + // as RFC3339 string in sessions.metadata JSONB (flushed on next save). + if l.sessions != nil && req != nil && req.SessionKey != "" { + l.sessions.SetSessionMetadata(ctx, req.SessionKey, map[string]string{ + SessionMetaKeyLastCompactionAt: time.Now().UTC().Format(time.RFC3339), + }) + } return compacted, nil } } +// SessionMetaKeyLastCompactionAt is the sessions.metadata JSONB key used to +// record the RFC3339 timestamp of the most recent compaction. Exported so +// the web UI code path can read it back via GetSessionMetadata without +// duplicating the string. +const SessionMetaKeyLastCompactionAt = "last_compaction_at" + func (l *Loop) makeRunMemoryFlush() func(ctx context.Context, state *pipeline.RunState) error { return func(ctx context.Context, state *pipeline.RunState) error { settings := ResolveMemoryFlushSettings(l.compactionCfg) diff --git a/internal/config/config.go b/internal/config/config.go index 05b55fa9..37c2ce1d 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -186,6 +186,22 @@ type MemoryConfig struct { VectorWeight float64 `json:"vector_weight,omitempty"` // hybrid search vector weight (default 0.7) TextWeight float64 `json:"text_weight,omitempty"` // hybrid search FTS weight (default 0.3) MinScore float64 `json:"min_score,omitempty"` // minimum relevance score (default 0.35) + + // Dreaming configures the episodic → long-term consolidation worker. + // nil = use hardcoded defaults (threshold=5, debounce=10min, enabled). + Dreaming *DreamingConfig `json:"dreaming,omitempty"` +} + +// DreamingConfig controls per-agent behaviour of the consolidation dreaming +// worker (episodic summaries → long-term memory). Pointer fields allow partial +// overrides from JSONB to merge cleanly with defaults without clobbering +// unset values (e.g. leaving VerboseLog at its default when only Threshold is +// overridden). +type DreamingConfig struct { + Enabled *bool `json:"enabled,omitempty"` // default true (nil = enabled) + DebounceMs int `json:"debounce_ms,omitempty"` // min interval between runs per agent/user (default 600000 = 10 min) + Threshold int `json:"threshold,omitempty"` // min unpromoted entries before running (default 5) + VerboseLog *bool `json:"verbose_log,omitempty"` // log debounce/below-threshold skips at info level (default false) } // SandboxConfig configures Docker-based sandbox execution. diff --git a/internal/consolidation/dreaming_config.go b/internal/consolidation/dreaming_config.go new file mode 100644 index 00000000..afe6b503 --- /dev/null +++ b/internal/consolidation/dreaming_config.go @@ -0,0 +1,84 @@ +package consolidation + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// resolvedDreamingConfig is the fully-populated view of dreaming settings used +// inside Handle(). All fields are concrete (no pointers) so the worker can +// operate without further nil-checks. +type resolvedDreamingConfig struct { + Enabled bool + Debounce time.Duration + Threshold int + VerboseLog bool +} + +// defaultDreamingConfig returns hardcoded defaults that match the worker's +// legacy behaviour before per-agent overrides were added. These values are +// mirrored into the struct defaults in workers.go. +func defaultDreamingConfig() resolvedDreamingConfig { + return resolvedDreamingConfig{ + Enabled: true, + Debounce: dreamingDefaultDebounce, + Threshold: dreamingDefaultThreshold, + VerboseLog: false, + } +} + +// mergeDreamingConfig applies an agent-provided override onto the defaults. +// Nil pointer fields fall through to defaults; zero integers are treated as +// "unset" so a legacy empty JSONB still resolves to defaults. Negative values +// are clamped to defaults to defend against malformed config. +func mergeDreamingConfig(base resolvedDreamingConfig, override *config.DreamingConfig) resolvedDreamingConfig { + if override == nil { + return base + } + if override.Enabled != nil { + base.Enabled = *override.Enabled + } + if override.DebounceMs > 0 { + base.Debounce = time.Duration(override.DebounceMs) * time.Millisecond + } + if override.Threshold > 0 { + base.Threshold = override.Threshold + } + if override.VerboseLog != nil { + base.VerboseLog = *override.VerboseLog + } + return base +} + +// DreamingConfigResolver fetches per-agent dreaming config at event handling +// time. Implementations should be fast and non-blocking; the consolidation +// worker calls this on every event that passes the agent-ID parse. +type DreamingConfigResolver func(ctx context.Context, agentID string) *config.DreamingConfig + +// newAgentStoreResolver builds a DreamingConfigResolver that reads from the +// given AgentStore and extracts MemoryConfig.Dreaming. Returns nil if the +// store is nil so the worker can fall back to defaults. +func newAgentStoreResolver(agents store.AgentCRUDStore) DreamingConfigResolver { + if agents == nil { + return nil + } + return func(ctx context.Context, agentID string) *config.DreamingConfig { + id, err := uuid.Parse(agentID) + if err != nil { + return nil + } + ag, err := agents.GetByIDUnscoped(ctx, id) + if err != nil || ag == nil { + return nil + } + mc := ag.ParseMemoryConfig() + if mc == nil { + return nil + } + return mc.Dreaming + } +} diff --git a/internal/consolidation/dreaming_config_test.go b/internal/consolidation/dreaming_config_test.go new file mode 100644 index 00000000..d533ccd4 --- /dev/null +++ b/internal/consolidation/dreaming_config_test.go @@ -0,0 +1,197 @@ +package consolidation + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func boolPtr(b bool) *bool { return &b } + +func TestMergeDreamingConfigNilOverrideReturnsBase(t *testing.T) { + base := defaultDreamingConfig() + got := mergeDreamingConfig(base, nil) + if got != base { + t.Fatalf("nil override mutated base: got %+v, want %+v", got, base) + } +} + +func TestMergeDreamingConfigPartialOverride(t *testing.T) { + base := defaultDreamingConfig() + override := &config.DreamingConfig{ + Threshold: 3, + DebounceMs: 120_000, // 2 min + } + got := mergeDreamingConfig(base, override) + + if got.Threshold != 3 { + t.Errorf("Threshold = %d, want 3", got.Threshold) + } + if got.Debounce != 2*time.Minute { + t.Errorf("Debounce = %v, want 2m", got.Debounce) + } + // Enabled not set in override → stays at default. + if !got.Enabled { + t.Errorf("Enabled = false, want default true") + } +} + +func TestMergeDreamingConfigDisable(t *testing.T) { + base := defaultDreamingConfig() + override := &config.DreamingConfig{Enabled: boolPtr(false)} + got := mergeDreamingConfig(base, override) + if got.Enabled { + t.Errorf("Enabled = true, want false (override)") + } +} + +// TestMergeDreamingConfigVerboseLogNilPreservesDefault guards against the +// footgun where an override without VerboseLog silently reverts the base +// value. Pointer semantics require nil-check, not unconditional assignment. +func TestMergeDreamingConfigVerboseLogNilPreservesDefault(t *testing.T) { + base := defaultDreamingConfig() + base.VerboseLog = true // simulate an operator default of true + override := &config.DreamingConfig{Threshold: 2} + got := mergeDreamingConfig(base, override) + if !got.VerboseLog { + t.Errorf("VerboseLog = false, want true (nil override should preserve base)") + } +} + +func TestMergeDreamingConfigVerboseLogExplicitFalse(t *testing.T) { + base := defaultDreamingConfig() + base.VerboseLog = true + override := &config.DreamingConfig{VerboseLog: boolPtr(false)} + got := mergeDreamingConfig(base, override) + if got.VerboseLog { + t.Errorf("VerboseLog = true, want false (explicit override must apply)") + } +} + +func TestMergeDreamingConfigZeroFieldsIgnored(t *testing.T) { + // Zero values on non-pointer fields must not clobber defaults — this is + // critical for agents with an empty memory_config.dreaming JSONB object. + base := defaultDreamingConfig() + override := &config.DreamingConfig{} // all zero + got := mergeDreamingConfig(base, override) + if got.Threshold != base.Threshold { + t.Errorf("Threshold = %d, want base %d", got.Threshold, base.Threshold) + } + if got.Debounce != base.Debounce { + t.Errorf("Debounce = %v, want base %v", got.Debounce, base.Debounce) + } +} + +// TestDreamingWorkerHandleHonoursCustomThreshold verifies that a per-agent +// resolver lowering the threshold fires even when the worker's global default +// would have skipped the event. +func TestDreamingWorkerHandleHonoursCustomThreshold(t *testing.T) { + summaries := []store.EpisodicSummary{ + {ID: uuid.New(), Summary: "s1"}, + {ID: uuid.New(), Summary: "s2"}, + } + mockEpisodic := &mockEpisodicStore{ + countResult: 2, + unpromoted: summaries, + promoted: make(map[string]bool), + } + mockMemory := newMockMemoryStore() + mockProvider := &mockProvider{ + chatResp: &providers.ChatResponse{Content: "ok"}, + } + + worker := &dreamingWorker{ + episodicStore: mockEpisodic, + memoryStore: mockMemory, + provider: mockProvider, + model: "test", + threshold: 5, // global default says skip at count=2 + debounce: 1 * time.Second, + resolveConfig: func(_ context.Context, _ string) *config.DreamingConfig { + return &config.DreamingConfig{Threshold: 2} // per-agent override lowers it + }, + } + + err := worker.Handle(context.Background(), eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + TenantID: uuid.New().String(), + AgentID: "agent-123", + UserID: "user-123", + Payload: &eventbus.EpisodicCreatedPayload{}, + }) + if err != nil { + t.Fatalf("Handle failed: %v", err) + } + if len(mockEpisodic.promoted) != 2 { + t.Errorf("Expected 2 promotions with override threshold, got %d", len(mockEpisodic.promoted)) + } +} + +// TestDreamingWorkerHandleDisabledSkips verifies that Enabled=false prevents +// any store activity, regardless of threshold/debounce. +func TestDreamingWorkerHandleDisabledSkips(t *testing.T) { + mockEpisodic := &mockEpisodicStore{ + countResult: 100, // would normally trigger + promoted: make(map[string]bool), + } + + worker := &dreamingWorker{ + episodicStore: mockEpisodic, + threshold: 5, + debounce: 1 * time.Second, + resolveConfig: func(_ context.Context, _ string) *config.DreamingConfig { + return &config.DreamingConfig{Enabled: boolPtr(false)} + }, + } + + err := worker.Handle(context.Background(), eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + TenantID: uuid.New().String(), + AgentID: "agent-disabled", + UserID: "user-123", + Payload: &eventbus.EpisodicCreatedPayload{}, + }) + if err != nil { + t.Fatalf("Handle failed: %v", err) + } + if len(mockEpisodic.promoted) > 0 { + t.Errorf("Expected 0 promotions when disabled, got %d", len(mockEpisodic.promoted)) + } + // CountUnpromoted must also be skipped — it's a downstream read that + // dreaming workers should not perform when an agent opts out. + if mockEpisodic.countCalls > 0 { + t.Errorf("Expected 0 CountUnpromoted calls when disabled, got %d", mockEpisodic.countCalls) + } +} + +// TestDreamingWorkerHandleNilResolverUsesDefaults verifies backward +// compatibility — no resolver means the worker behaves exactly as before. +func TestDreamingWorkerHandleNilResolverUsesDefaults(t *testing.T) { + mockEpisodic := &mockEpisodicStore{countResult: 2} // below default 5 + worker := &dreamingWorker{ + episodicStore: mockEpisodic, + threshold: 5, + debounce: 1 * time.Second, + // resolveConfig: nil + } + err := worker.Handle(context.Background(), eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + TenantID: uuid.New().String(), + AgentID: "agent-123", + UserID: "user-123", + Payload: &eventbus.EpisodicCreatedPayload{}, + }) + if err != nil { + t.Fatalf("Handle failed: %v", err) + } + // Below threshold — no promotions expected (default behaviour preserved). + if len(mockEpisodic.promoted) > 0 { + t.Errorf("Expected no promotions below default threshold, got %d", len(mockEpisodic.promoted)) + } +} diff --git a/internal/consolidation/dreaming_scoring_test.go b/internal/consolidation/dreaming_scoring_test.go new file mode 100644 index 00000000..c42e3d58 --- /dev/null +++ b/internal/consolidation/dreaming_scoring_test.go @@ -0,0 +1,191 @@ +package consolidation + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/eventbus" + "github.com/nextlevelbuilder/goclaw/internal/providers" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// TestDreamingWorkerUsesScoredListing verifies the worker fetches via +// ListUnpromotedScored (not the legacy created_at ASC path) so recall-hot +// entries synthesise first. +func TestDreamingWorkerUsesScoredListing(t *testing.T) { + // Two unpromoted entries: one cold-but-old, one hot-recent. + now := time.Now().UTC() + lastRecall := now.Add(-1 * time.Hour) + summaries := []store.EpisodicSummary{ + {ID: uuid.New(), Summary: "hot", RecallCount: 5, RecallScore: 0.9, LastRecalledAt: &lastRecall, CreatedAt: now.Add(-2 * 24 * time.Hour)}, + {ID: uuid.New(), Summary: "cold", CreatedAt: now.Add(-5 * 24 * time.Hour)}, + } + // Custom scored-mock flagging whether the correct listing method fired. + mockEp := &scoredEpisodicMock{ + mockEpisodicStore: mockEpisodicStore{ + countResult: 5, + unpromoted: summaries, + promoted: make(map[string]bool), + }, + } + mockMem := newMockMemoryStore() + mockProv := &mockProvider{chatResp: &providers.ChatResponse{Content: "consolidated"}} + + worker := &dreamingWorker{ + episodicStore: mockEp, + memoryStore: mockMem, + provider: mockProv, + model: "test", + threshold: 5, + debounce: 1 * time.Second, + } + + err := worker.Handle(context.Background(), eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + TenantID: uuid.New().String(), + AgentID: "agent-scored", + UserID: "user-scored", + Payload: &eventbus.EpisodicCreatedPayload{}, + }) + if err != nil { + t.Fatalf("Handle failed: %v", err) + } + if !mockEp.scoredCalled { + t.Errorf("worker did not call ListUnpromotedScored (scoring bypass)") + } + if mockEp.legacyCalled { + t.Errorf("worker called legacy ListUnpromoted — should use scored path") + } + if len(mockEp.promoted) != 2 { + t.Errorf("expected 2 entries promoted, got %d", len(mockEp.promoted)) + } +} + +// TestDreamingWorkerFiltersBelowThreshold verifies that even when the agent +// has enough unpromoted entries to pass the count check, the recall-score +// filter can still skip synthesis if every entry is too weak. +func TestDreamingWorkerFiltersBelowThreshold(t *testing.T) { + now := time.Now().UTC() + // All entries recalled once (below MinRecallCount=2) with trivial score. + summaries := make([]store.EpisodicSummary, 5) + for i := range summaries { + summaries[i] = store.EpisodicSummary{ + ID: uuid.New(), + Summary: "weak", + CreatedAt: now.Add(-30 * 24 * time.Hour), + RecallCount: 1, + RecallScore: 0.1, + } + } + mockEp := &scoredEpisodicMock{ + mockEpisodicStore: mockEpisodicStore{ + countResult: 5, + unpromoted: summaries, + promoted: make(map[string]bool), + }, + } + mockMem := newMockMemoryStore() + mockProv := &mockProvider{chatResp: &providers.ChatResponse{Content: "noop"}} + + worker := &dreamingWorker{ + episodicStore: mockEp, + memoryStore: mockMem, + provider: mockProv, + model: "test", + threshold: 5, + debounce: 1 * time.Second, + } + + err := worker.Handle(context.Background(), eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + TenantID: uuid.New().String(), + AgentID: "agent-filter", + UserID: "user-filter", + Payload: &eventbus.EpisodicCreatedPayload{}, + }) + if err != nil { + t.Fatalf("Handle failed: %v", err) + } + if len(mockEp.promoted) > 0 { + t.Errorf("expected all entries filtered, got %d promoted", len(mockEp.promoted)) + } + if len(mockMem.docs) > 0 { + t.Errorf("expected no synthesis document, got %d", len(mockMem.docs)) + } +} + +// TestDreamingWorkerFilterEmptyStampsDebounce guards against the starvation +// loop: when all entries fail the recall threshold filter we still bump +// lastRun, otherwise every subsequent episodic.created event re-runs the +// full scoring pipeline in a tight loop. See code review note P10.1. +func TestDreamingWorkerFilterEmptyStampsDebounce(t *testing.T) { + now := time.Now().UTC() + summaries := make([]store.EpisodicSummary, 5) + for i := range summaries { + summaries[i] = store.EpisodicSummary{ + ID: uuid.New(), + Summary: "weak", + CreatedAt: now.Add(-30 * 24 * time.Hour), + RecallCount: 1, // below MinRecallCount=2 → filtered + RecallScore: 0.05, + } + } + mockEp := &scoredEpisodicMock{ + mockEpisodicStore: mockEpisodicStore{ + countResult: 5, + unpromoted: summaries, + promoted: make(map[string]bool), + }, + } + worker := &dreamingWorker{ + episodicStore: mockEp, + memoryStore: newMockMemoryStore(), + provider: &mockProvider{chatResp: &providers.ChatResponse{Content: "noop"}}, + model: "test", + threshold: 5, + debounce: 10 * time.Minute, // realistic + } + + ev := eventbus.DomainEvent{ + Type: eventbus.EventEpisodicCreated, + TenantID: uuid.New().String(), + AgentID: "agent-loop", + UserID: "user-loop", + Payload: &eventbus.EpisodicCreatedPayload{}, + } + + // First event: filter drops everything → no promotions. + if err := worker.Handle(context.Background(), ev); err != nil { + t.Fatalf("first Handle: %v", err) + } + firstCountCalls := mockEp.countCalls + + // Second event immediately after: debounce should skip BEFORE CountUnpromoted. + if err := worker.Handle(context.Background(), ev); err != nil { + t.Fatalf("second Handle: %v", err) + } + if mockEp.countCalls != firstCountCalls { + t.Errorf("debounce not stamped on filter-empty skip — CountUnpromoted called %d times, want %d", + mockEp.countCalls, firstCountCalls) + } +} + +// scoredEpisodicMock wraps mockEpisodicStore with tracking flags for the +// two ListUnpromoted* methods so tests can assert which path the worker used. +type scoredEpisodicMock struct { + mockEpisodicStore + scoredCalled bool + legacyCalled bool +} + +func (m *scoredEpisodicMock) ListUnpromoted(ctx context.Context, agentID, userID string, limit int) ([]store.EpisodicSummary, error) { + m.legacyCalled = true + return m.mockEpisodicStore.ListUnpromoted(ctx, agentID, userID, limit) +} + +func (m *scoredEpisodicMock) ListUnpromotedScored(ctx context.Context, agentID, userID string, limit int) ([]store.EpisodicSummary, error) { + m.scoredCalled = true + return m.mockEpisodicStore.ListUnpromotedScored(ctx, agentID, userID, limit) +} diff --git a/internal/consolidation/dreaming_worker.go b/internal/consolidation/dreaming_worker.go index 519363b7..7eaf46de 100644 --- a/internal/consolidation/dreaming_worker.go +++ b/internal/consolidation/dreaming_worker.go @@ -27,10 +27,56 @@ type dreamingWorker struct { episodicStore store.EpisodicStore memoryStore store.MemoryStore provider providers.Provider - model string // LLM model for synthesis - threshold int // min unpromoted entries before running - debounce time.Duration // min interval between runs per agent/user - lastRun sync.Map // key: "agentID:userID" → time.Time + model string // LLM model for synthesis + + // threshold/debounce are the global defaults. Per-agent overrides come + // from resolveConfig which reads the agent's MemoryConfig.Dreaming JSONB. + threshold int + debounce time.Duration + resolveConfig DreamingConfigResolver + + lastRun sync.Map // key: "agentID:userID" → time.Time +} + +// formatEntryForSynthesis renders a single episodic entry with recall +// metadata for the LLM synthesis prompt. Entries with recall signal are +// tagged so the LLM can weight them higher; unrecalled entries pass through +// as plain summaries. +func formatEntryForSynthesis(e store.EpisodicSummary) string { + if e.RecallCount == 0 { + return e.Summary + } + lastRecall := "never" + if e.LastRecalledAt != nil { + lastRecall = e.LastRecalledAt.Format("Jan 2") + } + return fmt.Sprintf("[recalled %dx, last: %s] %s", e.RecallCount, lastRecall, e.Summary) +} + +// logSkip emits dreaming skip reasons at debug level by default, elevating +// to info when verbose logging is enabled via per-agent config. +func logSkip(verbose bool, msg string, args ...any) { + if verbose { + slog.Info(msg, args...) + return + } + slog.Debug(msg, args...) +} + +// effectiveConfig merges the worker's defaults with any per-agent override. +// Centralised so tests can reason about the precedence in one place. +func (w *dreamingWorker) effectiveConfig(ctx context.Context, agentID string) resolvedDreamingConfig { + base := defaultDreamingConfig() + if w.threshold > 0 { + base.Threshold = w.threshold + } + if w.debounce > 0 { + base.Debounce = w.debounce + } + if w.resolveConfig == nil { + return base + } + return mergeDreamingConfig(base, w.resolveConfig(ctx, agentID)) } // Handle processes an episodic.created event for the dreaming pipeline. @@ -45,41 +91,55 @@ func (w *dreamingWorker) Handle(ctx context.Context, event eventbus.DomainEvent) agentID := event.AgentID userID := event.UserID + // Resolve per-agent config (threshold/debounce/enabled). Falls back to + // struct defaults when no override is stored on the agent. + cfg := w.effectiveConfig(ctx, agentID) + if !cfg.Enabled { + if cfg.VerboseLog { + slog.Info("dreaming: disabled for agent", "agent", agentID, "user", userID) + } + return nil + } + // Debounce: skip if ran recently for this pair. key := agentID + ":" + userID - debounce := w.debounce - if debounce == 0 { - debounce = dreamingDefaultDebounce - } if v, ok := w.lastRun.Load(key); ok { - if last, ok := v.(time.Time); ok && time.Since(last) < debounce { - slog.Debug("dreaming: debounce skip", "agent", agentID, "user", userID) + if last, ok := v.(time.Time); ok && time.Since(last) < cfg.Debounce { + logSkip(cfg.VerboseLog, "dreaming: debounce skip", "agent", agentID, "user", userID) return nil } } // Count unpromoted entries; skip if below threshold. - threshold := w.threshold - if threshold <= 0 { - threshold = dreamingDefaultThreshold - } count, err := w.episodicStore.CountUnpromoted(ctx, agentID, userID) if err != nil { slog.Warn("dreaming: count unpromoted failed", "err", err, "agent", agentID) return nil } - if count < threshold { - slog.Debug("dreaming: below threshold", "count", count, "threshold", threshold, "agent", agentID) + if count < cfg.Threshold { + logSkip(cfg.VerboseLog, "dreaming: below threshold", "count", count, "threshold", cfg.Threshold, "agent", agentID) return nil } - // Fetch unpromoted entries. - entries, err := w.episodicStore.ListUnpromoted(ctx, agentID, userID, dreamingFetchLimit) + // Fetch unpromoted entries ordered by recall_score DESC (Phase 10). Falls + // back to created_at ASC for ties so agents with no recall history still + // get oldest-first behaviour. + entries, err := w.episodicStore.ListUnpromotedScored(ctx, agentID, userID, dreamingFetchLimit) if err != nil { slog.Warn("dreaming: list unpromoted failed", "err", err, "agent", agentID) return nil } + // Filter by recall-score thresholds before synthesis so weak entries + // don't burn LLM tokens. Never-recalled entries bypass the count minimum + // via the freshness component in ComputeRecallScore. + entries = filterByRecallThresholds(entries, defaultRecallThresholds(), time.Now().UTC()) if len(entries) == 0 { + // Stamp lastRun even on empty-filter skips — otherwise every subsequent + // episodic.created event re-runs CountUnpromoted + ListUnpromotedScored + // + filter in a tight loop until the agent accumulates fresh content. + // See code review note P10.1. + w.lastRun.Store(key, time.Now()) + logSkip(cfg.VerboseLog, "dreaming: all entries below recall thresholds", "agent", agentID, "user", userID) return nil } @@ -119,10 +179,12 @@ func (w *dreamingWorker) Handle(ctx context.Context, event eventbus.DomainEvent) } // synthesize calls the LLM to extract long-term facts from session summaries. +// Each entry is annotated with its recall metadata so the LLM can weight +// frequently-recalled memories higher during synthesis. func (w *dreamingWorker) synthesize(ctx context.Context, entries []store.EpisodicSummary) (string, error) { summaries := make([]string, len(entries)) for i, e := range entries { - summaries[i] = e.Summary + summaries[i] = formatEntryForSynthesis(e) } body := strings.Join(summaries, "\n---\n") diff --git a/internal/consolidation/scoring.go b/internal/consolidation/scoring.go new file mode 100644 index 00000000..1e74245a --- /dev/null +++ b/internal/consolidation/scoring.go @@ -0,0 +1,128 @@ +package consolidation + +import ( + "math" + "time" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// Scoring constants. Chosen to match the plan's 4-component formula (plan has +// 6 components but GoClaw lacks per-episode conceptual + consolidation data). +// +// score = 0.30*frequency + 0.35*relevance + 0.20*recency + 0.15*freshness +// +// halfLifeDays=14 matches typical short-term memory retention windows; tuning +// should be informed by operator feedback once real recall data accumulates. +const ( + recallWeightFrequency = 0.30 + recallWeightRelevance = 0.35 + recallWeightRecency = 0.20 + recallWeightFreshness = 0.15 + recallHalfLifeDays = 14.0 + // frequencyCap: recall_count above this yields max frequency component + // (avoids runaway-frequency bias for hot entries). + recallFrequencyCap = 10 +) + +// recallThresholds holds the minimum scoring cutoffs applied to +// ListUnpromotedScored results before LLM synthesis. Chosen well below the +// TS reference (0.75 / 3) because GoClaw boots with zero recall data — +// early-life agents must still reach synthesis via the freshness component. +type recallThresholds struct { + MinScore float64 + MinRecallCount int +} + +func defaultRecallThresholds() recallThresholds { + return recallThresholds{MinScore: 0.2, MinRecallCount: 2} +} + +// ComputeRecallScore computes the weighted recall score for an episodic entry. +// Pure function — no IO, no randomness — so the dreaming worker can compute +// scores on demand instead of materialising them into the DB on every event. +// +// Returned value is in [0, 1+ε]. In practice results cluster in [0, 0.8]; +// thresholds in defaultRecallThresholds are calibrated to that range. +func ComputeRecallScore(entry store.EpisodicSummary, now time.Time) float64 { + // 1. Frequency (30%): log-normalised recall count. Log1p smooths the + // curve so 0 → 0, 1 → 0.29, 10 → 1.0; anything above cap stays at 1. + freq := math.Log1p(float64(entry.RecallCount)) / math.Log1p(recallFrequencyCap) + if freq > 1.0 { + freq = 1.0 + } + + // 2. Relevance (35%): running average of memory_search hit scores. + // Already lives in recall_score column and is clamped server-side to [0,1] + // by RecordRecall callers. Defensive clamp here for robustness. + rel := entry.RecallScore + if rel < 0 { + rel = 0 + } else if rel > 1 { + rel = 1 + } + + // Exponential decay shared by recency + freshness components. + // lambda = ln(2) / halfLife → score halves every `recallHalfLifeDays` days. + lambda := math.Ln2 / recallHalfLifeDays + + // 3. Recency (20%): decay from LastRecalledAt (falls back to CreatedAt + // for entries that have never been searched). Entries recalled today + // score ~1.0; 14 days ago → 0.5; 28 days ago → 0.25. + lastRef := entry.CreatedAt + if entry.LastRecalledAt != nil { + lastRef = *entry.LastRecalledAt + } + recency := decayFrom(lastRef, now, lambda) + + // 4. Freshness (15%): decay from CreatedAt regardless of recall history. + // Keeps brand-new entries eligible even with zero recall — prevents cold + // start from starving the synthesis pipeline. + freshness := decayFrom(entry.CreatedAt, now, lambda) + + return recallWeightFrequency*freq + + recallWeightRelevance*rel + + recallWeightRecency*recency + + recallWeightFreshness*freshness +} + +// decayFrom returns exp(-lambda * ageDays). Negative ages (clock skew, +// future-dated rows) clamp to 1.0 so the test fixtures are predictable. +func decayFrom(then, now time.Time, lambda float64) float64 { + if then.IsZero() || !then.Before(now) { + return 1.0 + } + ageDays := now.Sub(then).Hours() / 24 + return math.Exp(-lambda * ageDays) +} + +// filterByRecallThresholds drops entries whose recall signal is too weak +// to justify LLM synthesis. Pure function so the dreaming worker can +// sort-then-filter in a single pass. +// +// Bootstrap-friendly rules: +// - Never-recalled entries (RecallCount == 0) bypass BOTH thresholds. +// Freshness + recency keep them eligible so new agents can synthesise +// their earliest sessions without any memory_search activity. +// - Entries that HAVE been recalled must clear both MinRecallCount and +// MinScore. If an entry has seen 1 weak hit, it's filtered. +func filterByRecallThresholds(entries []store.EpisodicSummary, th recallThresholds, now time.Time) []store.EpisodicSummary { + if len(entries) == 0 { + return entries + } + kept := entries[:0] + for _, e := range entries { + if e.RecallCount == 0 { + kept = append(kept, e) + continue + } + if e.RecallCount < th.MinRecallCount { + continue + } + if ComputeRecallScore(e, now) < th.MinScore { + continue + } + kept = append(kept, e) + } + return kept +} diff --git a/internal/consolidation/scoring_test.go b/internal/consolidation/scoring_test.go new file mode 100644 index 00000000..8e3e4bb9 --- /dev/null +++ b/internal/consolidation/scoring_test.go @@ -0,0 +1,160 @@ +package consolidation + +import ( + "math" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// floatEq checks two floats are equal within epsilon for the scoring assertions. +func floatEq(a, b, eps float64) bool { return math.Abs(a-b) < eps } + +func TestComputeRecallScoreZeroRecallFreshEntry(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entry := store.EpisodicSummary{ + ID: uuid.New(), + CreatedAt: now, // brand new, zero recalls + } + // freq=0, rel=0, recency=1.0, freshness=1.0 + // 0.30*0 + 0.35*0 + 0.20*1 + 0.15*1 = 0.35 + score := ComputeRecallScore(entry, now) + if !floatEq(score, 0.35, 0.001) { + t.Errorf("score = %f, want ~0.35", score) + } +} + +func TestComputeRecallScoreOldUnrecalled(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entry := store.EpisodicSummary{ + CreatedAt: now.AddDate(0, 0, -28), // 28 days old, never recalled + } + // recency = freshness = exp(-ln2/14 * 28) = exp(-ln2*2) = 0.25 + // score = 0 + 0 + 0.2*0.25 + 0.15*0.25 = 0.0875 + score := ComputeRecallScore(entry, now) + if !floatEq(score, 0.0875, 0.001) { + t.Errorf("score = %f, want ~0.0875", score) + } +} + +func TestComputeRecallScoreExponentialDecayHalfLife(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + // Entry recalled exactly 14 days ago (one half-life). Recency should halve. + lastRecall := now.AddDate(0, 0, -14) + entry := store.EpisodicSummary{ + CreatedAt: lastRecall, // created at same moment it was last recalled + RecallCount: 1, + RecallScore: 1.0, + LastRecalledAt: &lastRecall, + } + // freq = log1p(1) / log1p(10) = 0.693/2.398 ≈ 0.289 + // rel = 1.0 + // recency = freshness = exp(-ln2) = 0.5 + // score = 0.30*0.289 + 0.35*1.0 + 0.20*0.5 + 0.15*0.5 = 0.0867 + 0.35 + 0.175 = 0.6117 + score := ComputeRecallScore(entry, now) + if !floatEq(score, 0.6117, 0.001) { + t.Errorf("score = %f, want ~0.6117", score) + } +} + +func TestComputeRecallScoreHighRecallSignal(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entry := store.EpisodicSummary{ + CreatedAt: now.AddDate(0, 0, -3), // 3 days old + RecallCount: 10, + RecallScore: 0.9, + LastRecalledAt: &now, // recalled today + } + // freq ≈ 1.0 (cap) + // rel = 0.9 + // recency ≈ 1.0 (lastRef=now) + // freshness = exp(-ln2/14 * 3) ≈ 0.862 + // score ≈ 0.30 + 0.315 + 0.20 + 0.1293 ≈ 0.944 + score := ComputeRecallScore(entry, now) + if score < 0.9 { + t.Errorf("score = %f, want >0.9 (hot entry)", score) + } +} + +func TestComputeRecallScoreClampsRelevance(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + // Malformed entry with out-of-range recall_score — scoring must clamp. + entry := store.EpisodicSummary{ + CreatedAt: now, + RecallCount: 1, + RecallScore: 2.5, // bug: > 1.0 + } + score := ComputeRecallScore(entry, now) + if score > 1.1 { + t.Errorf("score = %f exceeds reasonable bounds (relevance should clamp)", score) + } +} + +func TestFilterByRecallThresholdsFreshBypass(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entries := []store.EpisodicSummary{ + // Never recalled, 28 days old → low score but bypasses via RecallCount=0. + {ID: uuid.New(), CreatedAt: now.AddDate(0, 0, -28)}, + } + kept := filterByRecallThresholds(entries, recallThresholds{MinScore: 0.5, MinRecallCount: 2}, now) + if len(kept) != 1 { + t.Errorf("expected 1 kept (fresh bypass), got %d", len(kept)) + } +} + +func TestFilterByRecallThresholdsPreservesOrder(t *testing.T) { + // In-place filter must not scramble the caller's ORDER BY recall_score DESC. + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entries := []store.EpisodicSummary{ + {ID: uuid.MustParse("00000000-0000-0000-0000-000000000001"), CreatedAt: now, RecallCount: 10, RecallScore: 0.9, LastRecalledAt: &now}, + {ID: uuid.MustParse("00000000-0000-0000-0000-000000000002"), CreatedAt: now, RecallCount: 1, RecallScore: 0.1}, // filtered + {ID: uuid.MustParse("00000000-0000-0000-0000-000000000003"), CreatedAt: now, RecallCount: 5, RecallScore: 0.7, LastRecalledAt: &now}, + } + kept := filterByRecallThresholds(entries, defaultRecallThresholds(), now) + if len(kept) != 2 { + t.Fatalf("len=%d, want 2", len(kept)) + } + if kept[0].ID.String() != "00000000-0000-0000-0000-000000000001" || + kept[1].ID.String() != "00000000-0000-0000-0000-000000000003" { + t.Errorf("order corrupted: %v", []string{kept[0].ID.String(), kept[1].ID.String()}) + } +} + +func TestFilterByRecallThresholdsDropsWeakRecalled(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entries := []store.EpisodicSummary{ + // Recalled once — below MinRecallCount=2 → filtered. + {ID: uuid.New(), CreatedAt: now, RecallCount: 1, RecallScore: 0.9}, + // Recalled 3 times with good score — kept. + {ID: uuid.New(), CreatedAt: now, RecallCount: 3, RecallScore: 0.8, LastRecalledAt: &now}, + // Never recalled — kept via freshness. + {ID: uuid.New(), CreatedAt: now}, + } + kept := filterByRecallThresholds(entries, defaultRecallThresholds(), now) + if len(kept) != 2 { + t.Errorf("expected 2 kept, got %d", len(kept)) + } +} + +func TestFormatEntryForSynthesisAnnotatesRecalled(t *testing.T) { + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + entry := store.EpisodicSummary{ + Summary: "test session", + RecallCount: 5, + LastRecalledAt: &now, + } + out := formatEntryForSynthesis(entry) + if !strings.Contains(out, "recalled 5x") || !strings.Contains(out, "test session") { + t.Errorf("output missing metadata or summary: %q", out) + } +} + +func TestFormatEntryForSynthesisBypassesUnrecalled(t *testing.T) { + entry := store.EpisodicSummary{Summary: "fresh content"} + if out := formatEntryForSynthesis(entry); out != "fresh content" { + t.Errorf("unrecalled entry should pass through unchanged, got %q", out) + } +} diff --git a/internal/consolidation/workers.go b/internal/consolidation/workers.go index cf5c8b12..903cfc92 100644 --- a/internal/consolidation/workers.go +++ b/internal/consolidation/workers.go @@ -24,6 +24,10 @@ type ConsolidationDeps struct { Provider providers.Provider // for LLM summarization Model string Extractor EntityExtractor + // AgentStore is optional: when present, the dreaming worker reads + // per-agent overrides from MemoryConfig.Dreaming. If nil, the worker + // uses its built-in defaults for every agent. + AgentStore store.AgentCRUDStore } // Register wires all consolidation workers to the event bus. @@ -52,6 +56,7 @@ func Register(deps ConsolidationDeps) func() { model: deps.Model, threshold: dreamingDefaultThreshold, debounce: dreamingDefaultDebounce, + resolveConfig: newAgentStoreResolver(deps.AgentStore), } unsub1 := deps.EventBus.Subscribe(eventbus.EventSessionCompleted, episodic.Handle) diff --git a/internal/consolidation/workers_test.go b/internal/consolidation/workers_test.go index 6d816be3..14045dbb 100644 --- a/internal/consolidation/workers_test.go +++ b/internal/consolidation/workers_test.go @@ -15,16 +15,25 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/store" ) +// recallCall is a single RecordRecall invocation captured by the mock store +// so tests can assert the (id, score) stream forwarded by memory_search. +type recallCall struct { + ID string + Score float64 +} + // mockEpisodicStore implements store.EpisodicStore for testing. type mockEpisodicStore struct { - created []*store.EpisodicSummary - existsByID map[string]bool - unpromoted []store.EpisodicSummary - promoted map[string]bool - countResult int - pruneErr error - pruneCount int - mu sync.Mutex + created []*store.EpisodicSummary + existsByID map[string]bool + unpromoted []store.EpisodicSummary + promoted map[string]bool + countResult int + countCalls int // set by CountUnpromoted — used by disabled-skip tests + recallCalls []recallCall + pruneErr error + pruneCount int + mu sync.Mutex } func (m *mockEpisodicStore) Create(_ context.Context, ep *store.EpisodicSummary) error { @@ -68,6 +77,7 @@ func (m *mockEpisodicStore) PruneExpired(_ context.Context) (int, error) { func (m *mockEpisodicStore) CountUnpromoted(_ context.Context, _, _ string) (int, error) { m.mu.Lock() defer m.mu.Unlock() + m.countCalls++ return m.countResult, nil } @@ -77,6 +87,23 @@ func (m *mockEpisodicStore) ListUnpromoted(_ context.Context, _, _ string, _ int return m.unpromoted, nil } +// ListUnpromotedScored returns the same fixture as ListUnpromoted; tests that +// need to verify sort order override the method directly on a dedicated mock. +func (m *mockEpisodicStore) ListUnpromotedScored(_ context.Context, _, _ string, _ int) ([]store.EpisodicSummary, error) { + m.mu.Lock() + defer m.mu.Unlock() + return m.unpromoted, nil +} + +// RecordRecall appends the call into the recallCalls slice so tests can +// assert which episodic IDs received which score. +func (m *mockEpisodicStore) RecordRecall(_ context.Context, id string, score float64) error { + m.mu.Lock() + defer m.mu.Unlock() + m.recallCalls = append(m.recallCalls, recallCall{ID: id, Score: score}) + return nil +} + func (m *mockEpisodicStore) MarkPromoted(_ context.Context, ids []string) error { m.mu.Lock() defer m.mu.Unlock() diff --git a/internal/providers/anthropic.go b/internal/providers/anthropic.go index ad46661b..bafc267c 100644 --- a/internal/providers/anthropic.go +++ b/internal/providers/anthropic.go @@ -138,20 +138,29 @@ func (p *AnthropicProvider) Chat(ctx context.Context, req ChatRequest) (*ChatRes body := p.buildRequestBody(model, req, false) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(model, req)) - return RetryDo(ctx, p.retryConfig, func() (*ChatResponse, error) { + resp, err := RetryDo(ctx, p.retryConfig, func() (*ChatResponse, error) { respBody, err := p.doRequest(ctx, body) if err != nil { return nil, err } defer respBody.Close() - var resp anthropicResponse - if err := json.NewDecoder(respBody).Decode(&resp); err != nil { + var parsed anthropicResponse + if err := json.NewDecoder(respBody).Decode(&parsed); err != nil { return nil, fmt.Errorf("anthropic: decode response: %w", err) } - return p.parseResponse(&resp), nil + return p.parseResponse(&parsed), nil }) + // Drop user-visible reasoning after parsing for models flagged as leakers. + // Usage.ThinkingTokens and RawAssistantContent remain intact so billing + // and Anthropic tool-use thinking passback continue to work. + if resp != nil { + if strip, _ := req.Options[OptStripThinking].(bool); strip { + resp.Thinking = "" + } + } + return resp, err } func (p *AnthropicProvider) doRequest(ctx context.Context, body any) (io.ReadCloser, error) { diff --git a/internal/providers/anthropic_stream.go b/internal/providers/anthropic_stream.go index 7d6861de..128fa803 100644 --- a/internal/providers/anthropic_stream.go +++ b/internal/providers/anthropic_stream.go @@ -10,6 +10,11 @@ import ( func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, error) { model := resolveAnthropicModel(req.Model, p.defaultModel, p.registry) + // stripThinking: when true, drop reasoning tokens from user-visible output. + // Billing counters (thinkingChars → Usage.ThinkingTokens) and tool-passback + // RawAssistantContent remain untouched so billing and Anthropic's thinking + // block replay continue to work. + stripThinking, _ := req.Options[OptStripThinking].(bool) body := p.buildRequestBody(model, req, true) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(model, req)) @@ -80,10 +85,14 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC onChunk(StreamChunk{Content: ev.Delta.Text}) } case "thinking_delta": - result.Thinking += ev.Delta.Thinking + // Always count raw thinking bytes for billing estimation + // below, even when stripping user-visible output. thinkingChars += len(ev.Delta.Thinking) - if onChunk != nil { - onChunk(StreamChunk{Thinking: ev.Delta.Thinking}) + if !stripThinking { + result.Thinking += ev.Delta.Thinking + if onChunk != nil { + onChunk(StreamChunk{Thinking: ev.Delta.Thinking}) + } } case "input_json_delta": if len(result.ToolCalls) > 0 { diff --git a/internal/providers/codex.go b/internal/providers/codex.go index 66fa17d7..5d988144 100644 --- a/internal/providers/codex.go +++ b/internal/providers/codex.go @@ -118,6 +118,10 @@ func (p *CodexProvider) middlewareConfig(req ChatRequest) MiddlewareConfig { } func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, error) { + // stripThinking: drop reasoning summaries from ChatResponse.Thinking and + // onChunk callbacks. Usage.ThinkingTokens is still populated from the + // final response.usage payload (Phase 1 billing accuracy). + stripThinking, _ := req.Options[OptStripThinking].(bool) body := p.buildRequestBody(req, true) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(req)) @@ -142,7 +146,7 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk continue } - p.processSSEEvent(&event, result, toolCalls, streamState, onChunk) + p.processSSEEvent(&event, result, toolCalls, streamState, onChunk, stripThinking) } if err := sse.Err(); err != nil { @@ -181,7 +185,9 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk } // processSSEEvent handles a single SSE event during streaming. -func (p *CodexProvider) processSSEEvent(event *codexSSEEvent, result *ChatResponse, toolCalls map[string]*codexToolCallAcc, streamState *codexMessageStreamState, onChunk func(StreamChunk)) { +// stripThinking drops reasoning summaries from user-visible output while +// leaving billing counters (Usage.ThinkingTokens) untouched. +func (p *CodexProvider) processSSEEvent(event *codexSSEEvent, result *ChatResponse, toolCalls map[string]*codexToolCallAcc, streamState *codexMessageStreamState, onChunk func(StreamChunk), stripThinking bool) { switch event.Type { case "response.output_item.added": if event.Item != nil { @@ -228,11 +234,13 @@ func (p *CodexProvider) processSSEEvent(event *codexSSEEvent, result *ChatRespon } toolCalls[event.Item.ID] = acc case "reasoning": - for _, s := range event.Item.Summary { - if s.Text != "" { - result.Thinking += s.Text - if onChunk != nil { - onChunk(StreamChunk{Thinking: s.Text}) + if !stripThinking { + for _, s := range event.Item.Summary { + if s.Text != "" { + result.Thinking += s.Text + if onChunk != nil { + onChunk(StreamChunk{Thinking: s.Text}) + } } } } diff --git a/internal/providers/openai_chat.go b/internal/providers/openai_chat.go index daf8ee3d..ae9505f0 100644 --- a/internal/providers/openai_chat.go +++ b/internal/providers/openai_chat.go @@ -24,7 +24,16 @@ func (p *OpenAIProvider) Chat(ctx context.Context, req ChatRequest) (*ChatRespon if err != nil { if clamped := clampMaxTokensFromError(err, body); clamped { slog.Info("max_tokens clamped, retrying", "model", model, "limit", clampedLimit(body)) - return RetryDo(ctx, p.retryConfig, chatFn) + resp, err = RetryDo(ctx, p.retryConfig, chatFn) + } + } + + // Drop user-visible reasoning for models flagged as leakers (e.g. Kimi, + // DeepSeek-Reasoner). Usage.ThinkingTokens is preserved so billing stays + // correct (Phase 1 depends on this). + if resp != nil { + if strip, _ := req.Options[OptStripThinking].(bool); strip { + resp.Thinking = "" } } @@ -52,6 +61,9 @@ func (p *OpenAIProvider) chatRequestFn(ctx context.Context, body map[string]any) func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk func(StreamChunk)) (*ChatResponse, error) { model := p.resolveModel(req.Model) + // stripThinking suppresses user-visible reasoning while leaving + // Usage.ThinkingTokens untouched (the usage chunk below still records it). + stripThinking, _ := req.Options[OptStripThinking].(bool) body := p.buildRequestBody(model, req, true) body = ApplyMiddlewares(body, p.middlewares, p.middlewareConfig(model, req)) @@ -112,7 +124,7 @@ func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChun if reasoning == "" { reasoning = delta.Reasoning } - if reasoning != "" { + if reasoning != "" && !stripThinking { result.Thinking += reasoning if onChunk != nil { onChunk(StreamChunk{Thinking: reasoning}) diff --git a/internal/providers/reasoning_resolution.go b/internal/providers/reasoning_resolution.go index 0725208e..d02e05e1 100644 --- a/internal/providers/reasoning_resolution.go +++ b/internal/providers/reasoning_resolution.go @@ -17,9 +17,34 @@ type ReasoningDecision struct { KnownModel bool `json:"known_model,omitempty"` SupportedLevels []string `json:"supported_levels,omitempty"` UsedProviderDefault bool `json:"used_provider_default,omitempty"` + // StripThinking signals that ChatResponse.Thinking should be dropped by + // stream handlers (model leaks reasoning even when effort="off"). + // Usage.ThinkingTokens is preserved for billing accuracy. + StripThinking bool `json:"strip_thinking,omitempty"` } -func ResolveReasoningDecision(provider Provider, model, requestedEffort, fallback, source string) ReasoningDecision { +// modelLeaksReasoning returns true for models known to emit reasoning tokens +// into the user-visible content even when reasoning effort is "off". Stream +// handlers drop these from ChatResponse.Thinking to avoid leaking raw CoT to +// end users. Billing tokens (Usage.ThinkingTokens) are NOT affected. +func modelLeaksReasoning(model string) bool { + m := strings.ToLower(strings.TrimSpace(model)) + if m == "" { + return false + } + // Known leakers: Moonshot Kimi (any variant), DeepSeek-Reasoner family. + // Extend this list only after confirming the model does not honor effort="off". + return strings.Contains(m, "kimi") || strings.Contains(m, "deepseek-reasoner") +} + +func ResolveReasoningDecision(provider Provider, model, requestedEffort, fallback, source string) (out ReasoningDecision) { + // Post-process: when resolved effort is "off" on a known leaky model, + // mark the decision for downstream stream handlers to strip thinking output. + defer func() { + if out.EffectiveEffort == "off" && modelLeaksReasoning(model) { + out.StripThinking = true + } + }() decision := ReasoningDecision{ Source: normalizeReasoningSource(source), RequestedEffort: NormalizeReasoningEffort(requestedEffort), diff --git a/internal/providers/reasoning_strip_stream_test.go b/internal/providers/reasoning_strip_stream_test.go new file mode 100644 index 00000000..51fe3dc4 --- /dev/null +++ b/internal/providers/reasoning_strip_stream_test.go @@ -0,0 +1,102 @@ +package providers + +import ( + "context" + "testing" +) + +// TestAnthropicStream_StripThinkingDropsReasoningButKeepsUsage verifies that +// when OptStripThinking is set, thinking_delta events do NOT accumulate into +// ChatResponse.Thinking but Usage.ThinkingTokens is still estimated from the +// raw byte count (billing accuracy — Phase 1 depends on this). +func TestAnthropicStream_StripThinkingDropsReasoningButKeepsUsage(t *testing.T) { + events := []string{ + "event: message_start\n", + `data: {"message":{"usage":{"input_tokens":10}}}` + "\n\n", + + "event: content_block_start\n", + `data: {"index":0,"content_block":{"type":"thinking","thinking":""}}` + "\n\n", + + "event: content_block_delta\n", + `data: {"index":0,"delta":{"type":"thinking_delta","thinking":"raw chain of thought output that must be stripped"}}` + "\n\n", + + "event: content_block_stop\n", + "data: {}\n\n", + + "event: content_block_start\n", + `data: {"index":1,"content_block":{"type":"text","text":""}}` + "\n\n", + + "event: content_block_delta\n", + `data: {"index":1,"delta":{"type":"text_delta","text":"answer"}}` + "\n\n", + + "event: content_block_stop\n", + "data: {}\n\n", + + "event: message_delta\n", + `data: {"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":20}}` + "\n\n", + + "event: message_stop\n", + "data: {}\n\n", + } + server := newAnthropicSSEServer(t, events) + p := newTestAnthropicProvider(server.URL) + + var chunks []StreamChunk + req := ChatRequest{ + Model: "kimi-k2", + Messages: []Message{{Role: "user", Content: "hello"}}, + Options: map[string]any{ + OptStripThinking: true, + }, + } + result, err := p.ChatStream(context.Background(), req, func(chunk StreamChunk) { + chunks = append(chunks, chunk) + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result.Thinking != "" { + t.Errorf("Thinking = %q, want empty (stripped)", result.Thinking) + } + if result.Content != "answer" { + t.Errorf("Content = %q, want %q (content must still flow through)", result.Content, "answer") + } + // Usage.ThinkingTokens should still be counted from raw bytes: + // len("raw chain of thought output that must be stripped") = 49 → 49/4 = 12 tokens. + if result.Usage == nil || result.Usage.ThinkingTokens == 0 { + t.Errorf("Usage.ThinkingTokens = 0, want >0 (billing must still count stripped thinking)") + } + // No thinking chunks should have been emitted to the caller. + for _, c := range chunks { + if c.Thinking != "" { + t.Errorf("unexpected thinking chunk emitted while stripping: %q", c.Thinking) + } + } +} + +// TestAnthropicStream_StripThinkingFalseKeepsReasoning is the backward-compat +// smoke test: when the flag is absent, existing behaviour is unchanged. +func TestAnthropicStream_StripThinkingFalseKeepsReasoning(t *testing.T) { + events := []string{ + "event: content_block_delta\n", + `data: {"index":0,"delta":{"type":"thinking_delta","thinking":"visible thought"}}` + "\n\n", + "event: content_block_stop\n", + "data: {}\n\n", + "event: message_stop\n", + "data: {}\n\n", + } + server := newAnthropicSSEServer(t, events) + p := newTestAnthropicProvider(server.URL) + + req := ChatRequest{ + Model: "claude-sonnet-4-5-20250929", + Messages: []Message{{Role: "user", Content: "hello"}}, + } + result, err := p.ChatStream(context.Background(), req, nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result.Thinking != "visible thought" { + t.Errorf("Thinking = %q, want %q (no strip flag set)", result.Thinking, "visible thought") + } +} diff --git a/internal/providers/reasoning_strip_test.go b/internal/providers/reasoning_strip_test.go new file mode 100644 index 00000000..3d94291b --- /dev/null +++ b/internal/providers/reasoning_strip_test.go @@ -0,0 +1,67 @@ +package providers + +import ( + "testing" +) + +// TestResolveReasoningDecisionStripsForLeakyModels verifies that known-leaky +// models (Kimi, DeepSeek-Reasoner) flag StripThinking when effort resolves to +// "off", so stream handlers can drop raw CoT from user-visible output. +func TestResolveReasoningDecisionStripsForLeakyModels(t *testing.T) { + cases := []struct { + name string + model string + wantStrip bool + wantEffort string + }{ + {"kimi k2 default off", "kimi-k2", true, "off"}, + {"kimi thinking preview", "moonshot/kimi-k2-thinking", true, "off"}, + {"deepseek reasoner", "deepseek-reasoner", true, "off"}, + {"deepseek chat not flagged", "deepseek-chat", false, "off"}, + {"gpt-5.4 not flagged", "gpt-5.4", false, "off"}, + {"empty model", "", false, "off"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + decision := ResolveReasoningDecision( + testThinkingProvider{thinking: true}, + tc.model, "off", "downgrade", "reasoning", + ) + if decision.EffectiveEffort != tc.wantEffort { + t.Fatalf("EffectiveEffort = %q, want %q", decision.EffectiveEffort, tc.wantEffort) + } + if decision.StripThinking != tc.wantStrip { + t.Fatalf("StripThinking = %v, want %v", decision.StripThinking, tc.wantStrip) + } + }) + } +} + +// TestResolveReasoningDecisionNoStripWhenEffortActive ensures that when a user +// explicitly asks for reasoning on a leaky model, StripThinking stays false +// (user wants to see the reasoning output). +func TestResolveReasoningDecisionNoStripWhenEffortActive(t *testing.T) { + decision := ResolveReasoningDecision( + testThinkingProvider{thinking: true}, + "kimi-k2", "high", "downgrade", "reasoning", + ) + if decision.StripThinking { + t.Fatalf("StripThinking = true, want false (effort is active)") + } +} + +// TestModelLeaksReasoning spot-checks the model allowlist directly. +func TestModelLeaksReasoning(t *testing.T) { + leaky := []string{"kimi-k2", "KIMI-K2-Thinking", "deepseek-reasoner"} + for _, m := range leaky { + if !modelLeaksReasoning(m) { + t.Errorf("modelLeaksReasoning(%q) = false, want true", m) + } + } + safe := []string{"", "gpt-5.4", "claude-4-opus", "deepseek-chat", "o3-mini"} + for _, m := range safe { + if modelLeaksReasoning(m) { + t.Errorf("modelLeaksReasoning(%q) = true, want false", m) + } + } +} diff --git a/internal/providers/types.go b/internal/providers/types.go index 9d7d4a95..abf0fdf4 100644 --- a/internal/providers/types.go +++ b/internal/providers/types.go @@ -14,6 +14,10 @@ const ( OptReasoningEffort = "reasoning_effort" OptEnableThinking = "enable_thinking" OptThinkingBudget = "thinking_budget" + // OptStripThinking (bool) tells stream handlers to drop reasoning tokens + // from ChatResponse.Thinking and onChunk callbacks. Usage.ThinkingTokens + // and RawAssistantContent are preserved (billing + tool passback safety). + OptStripThinking = "strip_thinking" // Middleware-related options (Phase 2 will use these) OptServiceTier = "service_tier" diff --git a/internal/store/episodic_store.go b/internal/store/episodic_store.go index 44b4aab3..ce390cc6 100644 --- a/internal/store/episodic_store.go +++ b/internal/store/episodic_store.go @@ -10,20 +10,26 @@ import ( // EpisodicSummary represents a Tier 2 episodic memory entry. // Created from session summaries via the consolidation pipeline. type EpisodicSummary struct { - ID uuid.UUID `json:"id" db:"id"` - TenantID uuid.UUID `json:"tenant_id" db:"tenant_id"` - AgentID uuid.UUID `json:"agent_id" db:"agent_id"` - UserID string `json:"user_id" db:"user_id"` // string: chat-based IDs - SessionKey string `json:"session_key" db:"session_key"` - Summary string `json:"summary" db:"summary"` - KeyTopics []string `json:"key_topics" db:"key_topics"` - L0Abstract string `json:"l0_abstract" db:"l0_abstract"` // ~50 tokens, pre-computed - SourceType string `json:"source_type" db:"source_type"` // "session", "v2_daily", "manual" - SourceID string `json:"source_id" db:"source_id"` // dedup key - TurnCount int `json:"turn_count" db:"turn_count"` - TokenCount int `json:"token_count" db:"token_count"` - CreatedAt time.Time `json:"created_at" db:"created_at"` - ExpiresAt *time.Time `json:"expires_at,omitempty" db:"expires_at"` + ID uuid.UUID `json:"id" db:"id"` + TenantID uuid.UUID `json:"tenant_id" db:"tenant_id"` + AgentID uuid.UUID `json:"agent_id" db:"agent_id"` + UserID string `json:"user_id" db:"user_id"` // string: chat-based IDs + SessionKey string `json:"session_key" db:"session_key"` + Summary string `json:"summary" db:"summary"` + KeyTopics []string `json:"key_topics" db:"key_topics"` + L0Abstract string `json:"l0_abstract" db:"l0_abstract"` // ~50 tokens, pre-computed + SourceType string `json:"source_type" db:"source_type"` // "session", "v2_daily", "manual" + SourceID string `json:"source_id" db:"source_id"` // dedup key + TurnCount int `json:"turn_count" db:"turn_count"` + TokenCount int `json:"token_count" db:"token_count"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + ExpiresAt *time.Time `json:"expires_at,omitempty" db:"expires_at"` + + // Phase 10 — dreaming weighted scoring signals. Populated by + // EpisodicStore.RecordRecall; consumed by consolidation.ComputeRecallScore. + RecallCount int `json:"recall_count" db:"recall_count"` + RecallScore float64 `json:"recall_score" db:"recall_score"` // running average of memory_search hit scores + LastRecalledAt *time.Time `json:"last_recalled_at,omitempty" db:"last_recalled_at"` } // EpisodicSearchResult is a search hit with L0 summary. @@ -63,10 +69,18 @@ type EpisodicStore interface { // Promotion lifecycle (used by consolidation pipeline) // ListUnpromoted returns summaries not yet promoted to long-term memory, oldest first. ListUnpromoted(ctx context.Context, agentID, userID string, limit int) ([]EpisodicSummary, error) + // ListUnpromotedScored returns unpromoted summaries ordered by recall_score DESC + // (fallback: created_at ASC). Used by the dreaming worker to prioritise entries + // with stronger recall signals — see internal/consolidation/scoring.go. + ListUnpromotedScored(ctx context.Context, agentID, userID string, limit int) ([]EpisodicSummary, error) // MarkPromoted sets promoted_at=now() for the given IDs. MarkPromoted(ctx context.Context, ids []string) error // CountUnpromoted returns the count of unpromoted summaries for an agent/user. CountUnpromoted(ctx context.Context, agentID, userID string) (int, error) + // RecordRecall updates the per-episode recall signal after a memory_search hit. + // Implementations must increment recall_count, fold `score` into the running + // average stored in recall_score, and set last_recalled_at=NOW(). + RecordRecall(ctx context.Context, id string, score float64) error // Embedding SetEmbeddingProvider(provider EmbeddingProvider) diff --git a/internal/store/pg/episodic_search.go b/internal/store/pg/episodic_search.go index 70f77add..b9de7aa1 100644 --- a/internal/store/pg/episodic_search.go +++ b/internal/store/pg/episodic_search.go @@ -110,27 +110,15 @@ func mergeEpisodicScores(fts, vec []episodicScored, textWeight, vecWeight float6 return merged } -// scanEpisodic scans a single row into EpisodicSummary. +// scanEpisodic scans a single row into EpisodicSummary. Column order matches +// the SELECT list in PGEpisodicStore.Get (17 columns incl. recall signals). func scanEpisodic(row *sql.Row) (*store.EpisodicSummary, error) { var ep store.EpisodicSummary var topics pq.StringArray err := row.Scan(&ep.ID, &ep.TenantID, &ep.AgentID, &ep.UserID, &ep.SessionKey, &ep.Summary, &topics, &ep.TurnCount, &ep.TokenCount, - &ep.L0Abstract, &ep.SourceID, &ep.SourceType, &ep.CreatedAt, &ep.ExpiresAt) - if err != nil { - return nil, err - } - ep.KeyTopics = []string(topics) - return &ep, nil -} - -// scanEpisodicRow scans from sql.Rows (same fields as scanEpisodic). -func scanEpisodicRow(rows *sql.Rows) (*store.EpisodicSummary, error) { - var ep store.EpisodicSummary - var topics pq.StringArray - err := rows.Scan(&ep.ID, &ep.TenantID, &ep.AgentID, &ep.UserID, &ep.SessionKey, - &ep.Summary, &topics, &ep.TurnCount, &ep.TokenCount, - &ep.L0Abstract, &ep.SourceID, &ep.SourceType, &ep.CreatedAt, &ep.ExpiresAt) + &ep.L0Abstract, &ep.SourceID, &ep.SourceType, &ep.CreatedAt, &ep.ExpiresAt, + &ep.RecallCount, &ep.RecallScore, &ep.LastRecalledAt) if err != nil { return nil, err } diff --git a/internal/store/pg/episodic_summaries.go b/internal/store/pg/episodic_summaries.go index e7705094..3afacf70 100644 --- a/internal/store/pg/episodic_summaries.go +++ b/internal/store/pg/episodic_summaries.go @@ -68,7 +68,7 @@ func (s *PGEpisodicStore) Get(ctx context.Context, id string) (*store.EpisodicSu row := s.db.QueryRowContext(ctx, ` SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE id = $1 AND tenant_id = $2`, id, store.TenantIDFromContext(ctx)) return scanEpisodic(row) @@ -94,7 +94,8 @@ func (s *PGEpisodicStore) List(ctx context.Context, agentID, userID string, limi if userID != "" { q = `SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, + recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE agent_id = $1 AND user_id = $2 AND tenant_id = $5 ORDER BY created_at DESC LIMIT $3 OFFSET $4` @@ -102,7 +103,8 @@ func (s *PGEpisodicStore) List(ctx context.Context, agentID, userID string, limi } else { q = `SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, + recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE agent_id = $1 AND tenant_id = $4 ORDER BY created_at DESC LIMIT $2 OFFSET $3` @@ -194,19 +196,34 @@ func (s *PGEpisodicStore) PruneExpired(ctx context.Context) (int, error) { // ListUnpromoted returns episodic summaries not yet promoted to long-term memory, oldest first. // Scoped to agent/user within the caller's tenant. func (s *PGEpisodicStore) ListUnpromoted(ctx context.Context, agentID, userID string, limit int) ([]store.EpisodicSummary, error) { + return s.listUnpromoted(ctx, agentID, userID, limit, "created_at ASC") +} + +// ListUnpromotedScored returns unpromoted episodic summaries ordered by +// recall_score DESC (ties broken by created_at ASC so older entries with the +// same score synthesise first). Backed by `idx_episodic_recall_unpromoted`. +func (s *PGEpisodicStore) ListUnpromotedScored(ctx context.Context, agentID, userID string, limit int) ([]store.EpisodicSummary, error) { + return s.listUnpromoted(ctx, agentID, userID, limit, "recall_score DESC, created_at ASC") +} + +// listUnpromoted shares the query shape between the two ListUnpromoted* +// variants. `orderBy` is a static literal supplied by the caller — it is +// NEVER derived from user input, so the concatenation below is safe. +func (s *PGEpisodicStore) listUnpromoted(ctx context.Context, agentID, userID string, limit int, orderBy string) ([]store.EpisodicSummary, error) { if limit <= 0 { limit = 20 } tenantID := store.TenantIDFromContext(ctx) var rows []episodicSummaryRow - err := pkgSqlxDB.SelectContext(ctx, &rows, ` + query := ` SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, + recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE agent_id = $1 AND user_id = $2 AND tenant_id = $3 AND promoted_at IS NULL - ORDER BY created_at ASC LIMIT $4`, - agentID, userID, tenantID, limit) + ORDER BY ` + orderBy + ` LIMIT $4` + err := pkgSqlxDB.SelectContext(ctx, &rows, query, agentID, userID, tenantID, limit) if err != nil { return nil, fmt.Errorf("episodic list_unpromoted: %w", err) } @@ -217,6 +234,37 @@ func (s *PGEpisodicStore) ListUnpromoted(ctx context.Context, agentID, userID st return results, nil } +// RecordRecall increments recall_count, folds `score` into the running +// average stored in recall_score, and sets last_recalled_at=NOW(). Uses a +// single UPDATE so the row is rewritten atomically. +// +// The running-average expression reads the OLD values of recall_count / +// recall_score (Postgres evaluates the SET clause against the pre-update +// tuple), so the result is `(oldAvg*oldCount + newScore) / (oldCount+1)`. +func (s *PGEpisodicStore) RecordRecall(ctx context.Context, id string, score float64) error { + if id == "" { + return nil + } + // Clamp inputs so bad data from callers cannot corrupt the running average. + if score < 0 { + score = 0 + } else if score > 1 { + score = 1 + } + tenantID := store.TenantIDFromContext(ctx) + _, err := s.db.ExecContext(ctx, ` + UPDATE episodic_summaries + SET recall_count = recall_count + 1, + recall_score = (recall_score * recall_count + $1) / (recall_count + 1), + last_recalled_at = NOW() + WHERE id = $2 AND tenant_id = $3`, + score, id, tenantID) + if err != nil { + return fmt.Errorf("episodic record_recall: %w", err) + } + return nil +} + // MarkPromoted sets promoted_at=now() for the given episodic summary IDs. // Scoped to the caller's tenant for safety. func (s *PGEpisodicStore) MarkPromoted(ctx context.Context, ids []string) error { diff --git a/internal/store/pg/memory_scan_rows.go b/internal/store/pg/memory_scan_rows.go index 084460d7..80f0ccfa 100644 --- a/internal/store/pg/memory_scan_rows.go +++ b/internal/store/pg/memory_scan_rows.go @@ -102,34 +102,40 @@ func (r *scoredChunkRow) toScoredChunk() scoredChunk { // episodicSummaryRow is an sqlx scan struct for episodic_summaries SELECT queries. // Handles TEXT[] key_topics via pq.StringArray. type episodicSummaryRow struct { - ID string `db:"id"` - TenantID string `db:"tenant_id"` - AgentID string `db:"agent_id"` - UserID string `db:"user_id"` - SessionKey string `db:"session_key"` - Summary string `db:"summary"` - KeyTopics pq.StringArray `db:"key_topics"` - TurnCount int `db:"turn_count"` - TokenCount int `db:"token_count"` - L0Abstract string `db:"l0_abstract"` - SourceID string `db:"source_id"` - SourceType string `db:"source_type"` - CreatedAt time.Time `db:"created_at"` - ExpiresAt *time.Time `db:"expires_at"` + ID string `db:"id"` + TenantID string `db:"tenant_id"` + AgentID string `db:"agent_id"` + UserID string `db:"user_id"` + SessionKey string `db:"session_key"` + Summary string `db:"summary"` + KeyTopics pq.StringArray `db:"key_topics"` + TurnCount int `db:"turn_count"` + TokenCount int `db:"token_count"` + L0Abstract string `db:"l0_abstract"` + SourceID string `db:"source_id"` + SourceType string `db:"source_type"` + CreatedAt time.Time `db:"created_at"` + ExpiresAt *time.Time `db:"expires_at"` + RecallCount int `db:"recall_count"` + RecallScore float64 `db:"recall_score"` + LastRecalledAt *time.Time `db:"last_recalled_at"` } func (r *episodicSummaryRow) toEpisodicSummary() store.EpisodicSummary { ep := store.EpisodicSummary{ - UserID: r.UserID, - SessionKey: r.SessionKey, - Summary: r.Summary, - TurnCount: r.TurnCount, - TokenCount: r.TokenCount, - L0Abstract: r.L0Abstract, - SourceID: r.SourceID, - SourceType: r.SourceType, - CreatedAt: r.CreatedAt, - ExpiresAt: r.ExpiresAt, + UserID: r.UserID, + SessionKey: r.SessionKey, + Summary: r.Summary, + TurnCount: r.TurnCount, + TokenCount: r.TokenCount, + L0Abstract: r.L0Abstract, + SourceID: r.SourceID, + SourceType: r.SourceType, + CreatedAt: r.CreatedAt, + ExpiresAt: r.ExpiresAt, + RecallCount: r.RecallCount, + RecallScore: r.RecallScore, + LastRecalledAt: r.LastRecalledAt, } _ = ep.ID.Scan(r.ID) _ = ep.TenantID.Scan(r.TenantID) diff --git a/internal/store/sqlitestore/episodic.go b/internal/store/sqlitestore/episodic.go index 2456ff08..10fc16fa 100644 --- a/internal/store/sqlitestore/episodic.go +++ b/internal/store/sqlitestore/episodic.go @@ -69,7 +69,7 @@ func (s *SQLiteEpisodicStore) Get(ctx context.Context, id string) (*store.Episod row := s.db.QueryRowContext(ctx, ` SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE id = ? AND tenant_id = ?`, id, tenantID.String()) return scanSQLiteEpisodic(row) @@ -96,7 +96,7 @@ func (s *SQLiteEpisodicStore) List(ctx context.Context, agentID, userID string, if userID != "" { q = `SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE agent_id = ? AND user_id = ? AND tenant_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?` @@ -104,7 +104,7 @@ func (s *SQLiteEpisodicStore) List(ctx context.Context, agentID, userID string, } else { q = `SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE agent_id = ? AND tenant_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?` @@ -148,18 +148,33 @@ func (s *SQLiteEpisodicStore) PruneExpired(ctx context.Context) (int, error) { // ListUnpromoted returns summaries not yet promoted, oldest first. func (s *SQLiteEpisodicStore) ListUnpromoted(ctx context.Context, agentID, userID string, limit int) ([]store.EpisodicSummary, error) { + return s.listUnpromoted(ctx, agentID, userID, limit, "created_at ASC") +} + +// ListUnpromotedScored returns unpromoted summaries ordered by recall_score DESC +// (ties broken by created_at ASC). Backed by idx_episodic_recall_unpromoted +// (schema.sql). Mirrors PGEpisodicStore.ListUnpromotedScored. +func (s *SQLiteEpisodicStore) ListUnpromotedScored(ctx context.Context, agentID, userID string, limit int) ([]store.EpisodicSummary, error) { + return s.listUnpromoted(ctx, agentID, userID, limit, "recall_score DESC, created_at ASC") +} + +// listUnpromoted shares the query shape between the two ListUnpromoted* +// variants. `orderBy` is a static literal from the caller, never derived +// from user input, so the concatenation below is safe. +func (s *SQLiteEpisodicStore) listUnpromoted(ctx context.Context, agentID, userID string, limit int, orderBy string) ([]store.EpisodicSummary, error) { if limit <= 0 { limit = 20 } tenantID := tenantIDForInsert(ctx) - rows, err := s.db.QueryContext(ctx, ` + query := ` SELECT id, tenant_id, agent_id, user_id, session_key, summary, key_topics, turn_count, token_count, l0_abstract, source_id, source_type, - created_at, expires_at + created_at, expires_at, + recall_count, recall_score, last_recalled_at FROM episodic_summaries WHERE agent_id = ? AND user_id = ? AND tenant_id = ? AND promoted_at IS NULL - ORDER BY created_at ASC LIMIT ?`, - agentID, userID, tenantID.String(), limit) + ORDER BY ` + orderBy + ` LIMIT ?` + rows, err := s.db.QueryContext(ctx, query, agentID, userID, tenantID.String(), limit) if err != nil { return nil, fmt.Errorf("episodic list_unpromoted: %w", err) } @@ -167,6 +182,35 @@ func (s *SQLiteEpisodicStore) ListUnpromoted(ctx context.Context, agentID, userI return scanSQLiteEpisodicRows(rows) } +// RecordRecall increments recall_count, folds `score` into the running +// average stored in recall_score, and sets last_recalled_at. Atomic single +// UPDATE; the SET clause reads OLD row values (SQLite evaluates expressions +// against the pre-update tuple), so the running average is computed +// correctly in one statement. +func (s *SQLiteEpisodicStore) RecordRecall(ctx context.Context, id string, score float64) error { + if id == "" { + return nil + } + if score < 0 { + score = 0 + } else if score > 1 { + score = 1 + } + tenantID := tenantIDForInsert(ctx) + now := time.Now().UTC().Format(time.RFC3339Nano) + _, err := s.db.ExecContext(ctx, ` + UPDATE episodic_summaries + SET recall_count = recall_count + 1, + recall_score = (recall_score * recall_count + ?) / (recall_count + 1), + last_recalled_at = ? + WHERE id = ? AND tenant_id = ?`, + score, now, id, tenantID.String()) + if err != nil { + return fmt.Errorf("episodic record_recall: %w", err) + } + return nil +} + // MarkPromoted sets promoted_at=now() for the given IDs. // IDs are processed in chunks of 200 to avoid SQLite variable limit. func (s *SQLiteEpisodicStore) MarkPromoted(ctx context.Context, ids []string) error { @@ -211,18 +255,22 @@ func (s *SQLiteEpisodicStore) CountUnpromoted(ctx context.Context, agentID, user return count, nil } -// scanSQLiteEpisodic scans a single row into EpisodicSummary. +// scanSQLiteEpisodic scans a single row into EpisodicSummary. Column order +// matches the SELECT lists in Get / List / ListUnpromoted* (17 columns incl. +// Phase 10 recall signals). func scanSQLiteEpisodic(row *sql.Row) (*store.EpisodicSummary, error) { var ep store.EpisodicSummary var idStr, tenantStr, agentStr string var topicsBytes []byte var createdAt sqliteTime var expiresAt nullSqliteTime + var lastRecalledAt nullSqliteTime err := row.Scan( &idStr, &tenantStr, &agentStr, &ep.UserID, &ep.SessionKey, &ep.Summary, &topicsBytes, &ep.TurnCount, &ep.TokenCount, &ep.L0Abstract, &ep.SourceID, &ep.SourceType, - &createdAt, &expiresAt) + &createdAt, &expiresAt, + &ep.RecallCount, &ep.RecallScore, &lastRecalledAt) if errors.Is(err, sql.ErrNoRows) { return nil, nil } @@ -238,6 +286,10 @@ func scanSQLiteEpisodic(row *sql.Row) (*store.EpisodicSummary, error) { t := expiresAt.Time ep.ExpiresAt = &t } + if lastRecalledAt.Valid { + t := lastRecalledAt.Time + ep.LastRecalledAt = &t + } return &ep, nil } @@ -250,11 +302,13 @@ func scanSQLiteEpisodicRows(rows *sql.Rows) ([]store.EpisodicSummary, error) { var topicsBytes []byte var createdAt sqliteTime var expiresAt nullSqliteTime + var lastRecalledAt nullSqliteTime if err := rows.Scan( &idStr, &tenantStr, &agentStr, &ep.UserID, &ep.SessionKey, &ep.Summary, &topicsBytes, &ep.TurnCount, &ep.TokenCount, &ep.L0Abstract, &ep.SourceID, &ep.SourceType, - &createdAt, &expiresAt); err != nil { + &createdAt, &expiresAt, + &ep.RecallCount, &ep.RecallScore, &lastRecalledAt); err != nil { return nil, err } ep.ID, _ = uuid.Parse(idStr) @@ -266,6 +320,10 @@ func scanSQLiteEpisodicRows(rows *sql.Rows) ([]store.EpisodicSummary, error) { t := expiresAt.Time ep.ExpiresAt = &t } + if lastRecalledAt.Valid { + t := lastRecalledAt.Time + ep.LastRecalledAt = &t + } results = append(results, ep) } return results, rows.Err() diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index c9a790d5..5f664ecd 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -15,7 +15,7 @@ var schemaSQL string // SchemaVersion is the current SQLite schema version. // Bump this when adding new migration steps below. -const SchemaVersion = 12 +const SchemaVersion = 13 // migrations maps version → SQL to apply when upgrading FROM that version. // schema.sql always represents the LATEST full schema (for fresh DBs). @@ -366,6 +366,14 @@ WHERE a.deleted_at IS NULL AND NOT EXISTS (SELECT 1 FROM agent_context_files WHERE agent_id = a.id AND file_name = 'AGENTS_TASK.md'); DELETE FROM agent_context_files WHERE file_name = 'AGENTS_MINIMAL.md';`, + + // Version 12 → 13: Phase 10 dreaming weighted scoring signals on + // episodic_summaries. Mirrors PG migration 000045. + 12: `ALTER TABLE episodic_summaries ADD COLUMN recall_count INTEGER NOT NULL DEFAULT 0; +ALTER TABLE episodic_summaries ADD COLUMN recall_score REAL NOT NULL DEFAULT 0; +ALTER TABLE episodic_summaries ADD COLUMN last_recalled_at TEXT; +CREATE INDEX IF NOT EXISTS idx_episodic_recall_unpromoted ON episodic_summaries(agent_id, user_id, recall_score DESC) + WHERE promoted_at IS NULL;`, } // EnsureSchema creates tables if they don't exist and applies incremental migrations. diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index eedf56e0..c6de6b59 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -1414,7 +1414,12 @@ CREATE TABLE IF NOT EXISTS episodic_summaries ( token_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), expires_at TEXT, - promoted_at TEXT + promoted_at TEXT, + -- Phase 10: dreaming weighted scoring signals (running avg of memory_search + -- hit scores). See internal/consolidation/scoring.go::ComputeRecallScore. + recall_count INTEGER NOT NULL DEFAULT 0, + recall_score REAL NOT NULL DEFAULT 0, + last_recalled_at TEXT ); CREATE INDEX IF NOT EXISTS idx_episodic_agent_user ON episodic_summaries(agent_id, user_id); @@ -1423,6 +1428,8 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_episodic_source_dedup ON episodic_summarie WHERE source_id IS NOT NULL; CREATE INDEX IF NOT EXISTS idx_episodic_unpromoted ON episodic_summaries(agent_id, user_id, created_at) WHERE promoted_at IS NULL; +CREATE INDEX IF NOT EXISTS idx_episodic_recall_unpromoted ON episodic_summaries(agent_id, user_id, recall_score DESC) + WHERE promoted_at IS NULL; -- ============================================================ -- Table: agent_evolution_metrics (V3 self-evolution Stage 1) diff --git a/internal/tools/memory.go b/internal/tools/memory.go index a14e62a7..2b8aefe1 100644 --- a/internal/tools/memory.go +++ b/internal/tools/memory.go @@ -180,10 +180,47 @@ func (t *MemorySearchTool) Execute(ctx context.Context, args map[string]any) *Re // Record retrieval metric non-blocking (best-effort). t.recordRetrievalMetric(ctx, len(combined), episodicResults) + // Phase 10: update per-episode recall signals for dreaming weighted + // scoring. Fire-and-forget — recall tracking must never block the hot + // search path or surface errors to the agent loop. + t.recordEpisodicRecall(ctx, episodicResults) return NewResult(string(data)) } +// recordEpisodicRecall schedules a best-effort RecordRecall update per +// episodic hit so DreamingWorker can prioritise useful entries. Runs in a +// background goroutine bounded by a 5s timeout so slow DBs can't leak. +func (t *MemorySearchTool) recordEpisodicRecall(ctx context.Context, episodic []store.EpisodicSearchResult) { + if t.episodicStore == nil || len(episodic) == 0 { + return + } + tenantID := store.TenantIDFromContext(ctx) + if tenantID == uuid.Nil { + return + } + // Snapshot hits so the goroutine doesn't observe caller mutations. + hits := make([]store.EpisodicSearchResult, 0, len(episodic)) + for _, r := range episodic { + if r.EpisodicID == "" { + continue + } + hits = append(hits, r) + } + if len(hits) == 0 { + return + } + go func() { + bgCtx, cancel := context.WithTimeout(store.WithTenantID(context.Background(), tenantID), 5*time.Second) + defer cancel() + for _, r := range hits { + if err := t.episodicStore.RecordRecall(bgCtx, r.EpisodicID, r.Score); err != nil { + slog.Debug("memory.recall.record_failed", "episodic_id", r.EpisodicID, "error", err) + } + } + }() +} + // recordRetrievalMetric records a memory_search retrieval metric in a background goroutine. func (t *MemorySearchTool) recordRetrievalMetric(ctx context.Context, resultCount int, episodic []store.EpisodicSearchResult) { if t.metricsStore == nil { diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index c51fe279..beadd52e 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 44 +const RequiredSchemaVersion uint = 45 diff --git a/migrations/000045_episodic_recall_tracking.down.sql b/migrations/000045_episodic_recall_tracking.down.sql new file mode 100644 index 00000000..e519415c --- /dev/null +++ b/migrations/000045_episodic_recall_tracking.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS idx_episodic_recall_unpromoted; +ALTER TABLE episodic_summaries DROP COLUMN IF EXISTS last_recalled_at; +ALTER TABLE episodic_summaries DROP COLUMN IF EXISTS recall_score; +ALTER TABLE episodic_summaries DROP COLUMN IF EXISTS recall_count; diff --git a/migrations/000045_episodic_recall_tracking.up.sql b/migrations/000045_episodic_recall_tracking.up.sql new file mode 100644 index 00000000..49adab40 --- /dev/null +++ b/migrations/000045_episodic_recall_tracking.up.sql @@ -0,0 +1,15 @@ +-- Phase 10: dreaming weighted scoring. Track per-episode recall signals +-- that feed ComputeRecallScore() in internal/consolidation/scoring.go. +-- recall_score stores the running-average of search hit scores so the +-- dreaming worker can sort unpromoted entries by perceived usefulness +-- instead of strictly oldest-first. +ALTER TABLE episodic_summaries ADD COLUMN IF NOT EXISTS recall_count INT DEFAULT 0 NOT NULL; +ALTER TABLE episodic_summaries ADD COLUMN IF NOT EXISTS recall_score DOUBLE PRECISION DEFAULT 0 NOT NULL; +ALTER TABLE episodic_summaries ADD COLUMN IF NOT EXISTS last_recalled_at TIMESTAMPTZ; + +-- Partial index for DreamingWorker.ListUnpromotedScored — only touches +-- unpromoted rows and orders by recall_score DESC to match the primary +-- query shape: "top-N unpromoted with highest recall signal". +CREATE INDEX IF NOT EXISTS idx_episodic_recall_unpromoted + ON episodic_summaries(agent_id, user_id, recall_score DESC) + WHERE promoted_at IS NULL;