feat(ts-port): reasoning strip, dreaming config + weighted scoring

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.
2026-10-15 04:18:44 +00:00 · 2026-04-10 13:32:02 +07:00
1 parent dabb1eaa11
commit 4cf66eb379
@@ -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

@@ -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`
@@ -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)
@@ -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)
 	}()
 }
@@ -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)
@@ -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.
@@ -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
+	}
+}
@@ -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))
+	}
+}
@@ -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)
+}
@@ -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")

@@ -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
+}
@@ -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)
+	}
+}
@@ -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)
@@ -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()
@@ -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) {
@@ -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 {
@@ -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})
+							}
 						}
 					}
 				}
@@ -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})
@@ -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),
@@ -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")
+	}
+}
@@ -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)
+		}
+	}
+}
@@ -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"
@@ -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)
@@ -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
 	}
@@ -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 {
@@ -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)
@@ -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()
@@ -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.
@@ -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)
@@ -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 {
@@ -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
@@ -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;
@@ -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;