fix(memory): efficiency and correctness overhaul for extended + legacy memory

AGENTS.md

@@ -90,7 +90,7 @@ ReAct cycle: observe → think → act → repeat.
 - **Context-limit protection** — `trimToSurvival` drops oldest messages when approaching the model's context window, keeping the agent functional under extended sessions. Tool messages stay grouped with their parent assistant message.
 - **Interaction modes** — engaging (narrated), enhance (persistent), verbose (raw), off.
 - Max 300 iterations by default.
-- **Post-response async processing** — skill learning and episode extraction run in background goroutines, eliminating the hang after every `odek run`.
+- **Post-response async processing** — skill learning, episode extraction, and per-turn extended-memory extraction run in background goroutines tracked by `MemoryManager.RunBackground`; `Agent.Close` drains them via `WaitForBackground` (bounded, ~15s) so the work survives CLI exit without hanging `odek run`.
 - **Artifact-aware file search** — `search_files` and `multi_grep` skip build/artifact directories (`node_modules`, `vendor`, `.git`, `__pycache__`, `.venv`, etc.) automatically, reducing noise and speeding scans.
 - **Semantic session search** — the `session_search` tool uses go-vector RandomProjections + k-NN for semantic similarity search through session content, with a two-tier pipeline: vector index (fast, ~1ms) → deepSearch fallback (exhaustive, slower).
 - **Security-first defaults** — the latest hardening closes the high/medium/low findings tracked in `sec_findings.md`: default `non_interactive` is `deny`, project-level `odek.json` cannot redirect backends or hijack delivery, project-level sandbox knobs require explicit operator approval, `~/.odek` trust anchors are protected, WebSocket upgrades require a per-instance CSRF token, and all untrusted content is wrapped before reaching the model. See Security Architecture below for the full list.

cmd/odek/main.go

@@ -1493,9 +1493,11 @@ func run(args []string) error {
 	}
 
 	// ── Session end — extract episode if enough turns ──
-	// Run asynchronously so episode extraction does not delay process exit.
+	// Run in the background (tracked by the memory manager's WaitGroup) so
+	// episode extraction does not delay the response; Agent.Close drains it
+	// via WaitForBackground before process exit so it is not silently lost.
 	if mm := agent.Memory(); mm != nil && f.Session != nil && *f.Session && sessionID != "" {
-		go func() {
+		mm.RunBackground(func() {
 			store, err := session.NewStore()
 			if err == nil {
 				latest, err := store.Load(sessionID)
@@ -1505,7 +1507,7 @@ func run(args []string) error {
 					mm.OnSessionEndWithProvenance(latest.ID, latest.Turns, msgStrs, prov)
 				}
 			}
-		}()
+		})
 	}
 
 	// ── Delivery: send result to default channel ──
@@ -2472,13 +2474,15 @@ func continueCmd(args []string) error {
 	fmt.Fprintf(os.Stderr, "odek: session %s saved (%d turns)\n", sess.ID, sess.Turns+1)
 
 	// ── Session end — extract episode ──
-	// Run asynchronously so episode extraction does not delay process exit.
+	// Run in the background (tracked by the memory manager's WaitGroup) so
+	// episode extraction does not delay the response; Agent.Close drains it
+	// via WaitForBackground before process exit so it is not silently lost.
 	if mm := agent.Memory(); mm != nil {
-		go func() {
+		mm.RunBackground(func() {
 			msgStrs := makeSessionMessageStrings(sess)
 			prov := memory.DeriveProvenance(sess.Messages)
 			mm.OnSessionEndWithProvenance(sess.ID, sess.Turns+1, msgStrs, prov)
-		}()
+		})
 	}
 
 	return nil

cmd/odek/repl.go

@@ -292,17 +292,19 @@ func replCmd(args []string) error {
 	}
 
 	// Session end — extract episode if enough turns.
-	// Run asynchronously so episode extraction does not delay process exit.
+	// Run in the background (tracked by the memory manager's WaitGroup) so
+	// episode extraction does not delay REPL exit; the deferred Agent.Close
+	// drains it via WaitForBackground so it is not silently lost.
 	if mm := agent.Memory(); mm != nil {
-		go func() {
+		mm.RunBackground(func() {
 			messages := sess.GetMessages()
 			msgStrs := make([]string, 0, len(messages))
 			for _, m := range messages {
 				msgStrs = append(msgStrs, m.Role+": "+m.Content)
 			}
 			prov := memory.DeriveProvenance(messages)
 			mm.OnSessionEndWithProvenance(sess.ID, sess.Turns, msgStrs, prov)
-		}()
+		})
 	}
 
 	return nil

docs/EXTENDED_MEMORY.md

@@ -173,9 +173,9 @@ Extended Memory replaces episode-based recall with semantic search over atom vec
 2. Query the go-vector store for the top `semantic_search_top_k * semantic_search_overfetch` candidates.
 3. Drop tainted atoms and atoms below `semantic_search_min_score`.
 4. Compute a composite score: `0.6 * cosine_similarity + 0.4 * retention_score`.
-5. Optionally rerank the candidate set with the memory LLM.
-6. If `follow_up_anticipation_enabled` is true, generate predicted intents from the current message, recent messages, and the user model, and union their recall results with the literal-query results (including type-targeted recall for `convention`, `file`, and `error` atoms).
-7. Deduplicate, re-rank by retention score, and return the top-K atoms, bounded by `memory_budget_chars`.
+5. Optionally rerank the literal query's candidate set with the memory LLM (predicted-intent queries are not reranked).
+6. If `follow_up_anticipation_enabled` is true, generate predicted intents from the current message, recent messages, and the user model, and union their recall results with the literal-query results (including type-targeted recall for `convention`, `file`, and `error` atoms, filtered from the same candidate set).
+7. Deduplicate by atom ID (keeping the best composite score), re-rank by the composite score, and return the top-K atoms, bounded by `memory_budget_chars`.
 
 ### Ranking Formula
 
@@ -194,7 +194,7 @@ composite_score = 0.6 * cosine_similarity(query_vector, atom_vector)
                   + 0.4 * retention_score
 ```
 
-The final recall result is also bounded by `memory_budget_chars`. Tainted atoms are excluded regardless of score. When predictive intent recall is enabled, results from predicted intents are deduplicated with the literal-query results and then re-ranked by retention score before the top-K cut.
+The final recall result is also bounded by `memory_budget_chars`. Tainted atoms are excluded regardless of score. When predictive intent recall is enabled, results from predicted intents are deduplicated with the literal-query results (keeping the best composite score per atom ID) and then re-ranked by the composite score before the top-K cut.
 
 ## Predictive Recall
 

internal/embedding/embedder_test.go

@@ -280,3 +280,28 @@ func TestHTTPEmbedderCacheResetWhenFull(t *testing.T) {
 		t.Errorf("cache size = %d, want ≤ %d", size, maxEmbedCacheEntries)
 	}
 }
+
+// TestSharedFactory: the Shared factory returns ONE cache-warm instance for
+// stateless (HTTP) backends so consumers like episode dedup and the vector
+// index rebuild share the text→vector cache, but a FRESH instance per call
+// for corpus-fitted (RandomProjections) backends whose Fit state is per-corpus.
+func TestSharedFactory(t *testing.T) {
+	rpFactory := Shared(nil, 0)
+	if rpFactory() == rpFactory() {
+		t.Error("rp backend: Shared must return a fresh instance per call (per-corpus Fit state)")
+	}
+
+	srv, _, _ := mockEmbedServer(t)
+	httpFactory := Shared(&Config{Provider: "http", BaseURL: srv.URL + "/v1", Model: "mock-embed"}, 0)
+	first, second := httpFactory(), httpFactory()
+	if first != second {
+		t.Fatal("http backend: Shared must return the same cache-warm instance")
+	}
+	// Sanity: the shared instance works and caches across Fit calls.
+	if err := first.Fit([]string{"hello world"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := second.Embed("hello world"); err != nil {
+		t.Fatal(err)
+	}
+}

internal/embedding/embedding.go

@@ -156,6 +156,28 @@ func New(cfg *Config, rpDims int) TextEmbedder {
 	}
 }
 
+// Shared returns an embedder factory like New, except that for stateless
+// (HTTP) backends every call yields the SAME instance so its text→vector
+// cache is shared across consumers — e.g. episode dedup and the episode
+// vector-index rebuild then embed each corpus text once per process instead
+// of once per pass. Corpus-fitted backends (RandomProjections) still get a
+// FRESH instance per call: their Fit state is per-corpus, so sharing would
+// produce degenerate vectors. The HTTP embedder is internally mutex-guarded
+// and its Fit only warms the cache, so sharing it across goroutines is safe.
+func Shared(cfg *Config, rpDims int) func() TextEmbedder {
+	if emb := New(cfg, rpDims); isStateless(emb) {
+		return func() TextEmbedder { return emb }
+	}
+	return func() TextEmbedder { return New(cfg, rpDims) }
+}
+
+// isStateless reports whether the embedder is safe to share across consumers
+// (Fit does not capture corpus state).
+func isStateless(emb TextEmbedder) bool {
+	_, ok := emb.(*httpTextEmbedder)
+	return ok
+}
+
 // ── RandomProjections backend (default) ──────────────────────────────────────
 
 // rpTextEmbedder wraps go-vector RandomProjections behind TextEmbedder,

internal/loop/loop.go

@@ -559,6 +559,7 @@ func trimToSurvival(msgs []llm.Message) []llm.Message {
 // Run executes the loop for a given task and returns the final response.
 func (e *Engine) Run(ctx context.Context, task string) (string, error) {
 	e.memMsgIdx = -1
+	e.resetDedupKeys()
 	messages := []llm.Message{
 		{Role: "user", Content: task},
 	}
@@ -580,6 +581,7 @@ func (e *Engine) Run(ctx context.Context, task string) (string, error) {
 func (e *Engine) RunWithMessages(ctx context.Context, messages []llm.Message) (string, []llm.Message, error) {
 	// Reset token accounting for this run
 	e.memMsgIdx = -1
+	e.resetDedupKeys()
 	e.TotalInputTokens = 0
 	e.TotalOutputTokens = 0
 	e.TotalCacheCreationTokens = 0
@@ -588,6 +590,17 @@ func (e *Engine) RunWithMessages(ctx context.Context, messages []llm.Message) (s
 	return e.runLoop(ctx, messages)
 }
 
+// resetDedupKeys clears the per-message dedup keys so a repeated user
+// message in a later run (e.g. the REPL sending the same text twice)
+// re-triggers the memory hooks (user-message handler, skill loading,
+// episode recall, extended-memory recall).
+func (e *Engine) resetDedupKeys() {
+	e.lastUserMsg = ""
+	e.lastSkillMsg = ""
+	e.lastEpiMsg = ""
+	e.lastExtMsg = ""
+}
+
 // trustAllSetter is implemented by approvers (wsApprover, TelegramApprover)
 // whose tool-level prompts auto-pass while a batch approval grant is active.
 type trustAllSetter interface{ SetTrustAll(bool) }
@@ -661,8 +674,11 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
 		// Load relevant skills based on latest user input (once per message)
 		if e.skillLoader != nil {
 			if userMsg := lastUserMessage(messages); userMsg != "" && userMsg != e.lastSkillMsg {
+				// Assign the dedup key unconditionally — even when the loader
+				// finds no match — so a no-match doesn't re-run the (potentially
+				// slow) skill matcher on every remaining iteration of the turn.
+				e.lastSkillMsg = userMsg
 				if skillContext := e.skillLoader(userMsg); skillContext != "" {
-					e.lastSkillMsg = userMsg
 					// Inject skill context as a system message right before the user message.
 					// The skill manager gates NeedsReview/tainted skills, but we treat any
 					// loaded skill content as externally-sourced and wrap it with the
@@ -704,8 +720,11 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
 		// Only runs once per new user message (same dedup as skill loading).
 		if e.episodeCtx != nil {
 			if userMsg := lastUserMessage(messages); userMsg != "" && userMsg != e.lastEpiMsg {
+				// Assign the dedup key unconditionally — even when recall finds
+				// no match — so a no-match doesn't re-run the (potentially slow
+				// HTTP embed) episode search on every iteration of the turn.
+				e.lastEpiMsg = userMsg
 				if episodeContext := e.episodeCtx(userMsg); episodeContext != "" {
-					e.lastEpiMsg = userMsg
 					// Episode context comes from past session content and crosses the
 					// trust boundary; wrap it as untrusted before injecting.
 					wrappedContext := episodeContext

internal/loop/loop_test.go

@@ -750,6 +750,132 @@ func TestEngine_SkillLoader_CalledOncePerInput(t *testing.T) {
 	}
 }
 
+// twoIterationServer returns an httptest server whose first response requests
+// a tool call and whose second response is the final answer, forcing the loop
+// through exactly two iterations with the same user message.
+func twoIterationServer(t *testing.T) *httptest.Server {
+	t.Helper()
+	callCount := 0
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		callCount++
+		if callCount%2 == 1 {
+			// Odd iteration: request a tool call
+			fmt.Fprint(w, `{
+				"choices":[{
+					"message":{
+						"content":"Let me think.",
+						"tool_calls":[{
+							"id":"call_1",
+							"function":{
+								"name":"echo",
+								"arguments":"{}"
+							}
+						}]
+					}
+				}]
+			}`)
+		} else {
+			// Even iteration: final answer
+			fmt.Fprint(w, `{"choices":[{"message":{"content":"done"}}]}`)
+		}
+	}))
+	t.Cleanup(server.Close)
+	return server
+}
+
+func TestEngine_SkillLoader_NoMatchCalledOncePerInput(t *testing.T) {
+	// Regression: when the skill loader finds no match it must still record the
+	// dedup key, otherwise the matcher re-runs on every remaining iteration of
+	// the turn (each a potentially slow lookup).
+	skillLoadCount := 0
+	skillLoader := func(userInput string) string {
+		skillLoadCount++
+		return "" // no match
+	}
+
+	server := twoIterationServer(t)
+	echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"}
+	registry := tool.NewRegistry([]tool.Tool{echoTool})
+	client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0)
+	engine := New(client, registry, 10, "", nil, 0)
+	engine.SetSkillLoader(skillLoader)
+
+	if _, err := engine.Run(context.Background(), "do the task"); err != nil {
+		t.Fatalf("Run() error: %v", err)
+	}
+	if skillLoadCount != 1 {
+		t.Errorf("SkillLoader called %d times on a no-match, want 1 (dedup key must be set even when empty)", skillLoadCount)
+	}
+}
+
+func TestEngine_EpisodeCtx_NoMatchCalledOncePerInput(t *testing.T) {
+	// Regression: same dedup contract as the skill loader — a no-match episode
+	// recall must not re-run the (potentially slow) search every iteration.
+	episodeSearchCount := 0
+	episodeCtx := func(userInput string) string {
+		episodeSearchCount++
+		return "" // no match
+	}
+
+	server := twoIterationServer(t)
+	echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"}
+	registry := tool.NewRegistry([]tool.Tool{echoTool})
+	client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0)
+	engine := New(client, registry, 10, "", nil, 0)
+	engine.SetEpisodeContextFunc(episodeCtx)
+
+	if _, err := engine.Run(context.Background(), "do the task"); err != nil {
+		t.Fatalf("Run() error: %v", err)
+	}
+	if episodeSearchCount != 1 {
+		t.Errorf("episodeCtx called %d times on a no-match, want 1 (dedup key must be set even when empty)", episodeSearchCount)
+	}
+}
+
+func TestEngine_DedupKeysResetBetweenRuns(t *testing.T) {
+	// Regression: Run/RunWithMessages must reset the per-message dedup keys, so
+	// a REPL user sending the same text twice still gets the memory hooks
+	// (skill loading, episode recall, user-message handler) the second time.
+	skillLoadCount := 0
+	skillLoader := func(userInput string) string {
+		skillLoadCount++
+		return "skill ctx"
+	}
+	episodeSearchCount := 0
+	episodeCtx := func(userInput string) string {
+		episodeSearchCount++
+		return "episode ctx"
+	}
+	userMsgCount := 0
+	var userMsgHandler UserMessageHandler = func(ctx context.Context, msg string) {
+		userMsgCount++
+	}
+
+	server := twoIterationServer(t)
+	echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"}
+	registry := tool.NewRegistry([]tool.Tool{echoTool})
+	client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0)
+	engine := New(client, registry, 10, "", nil, 0)
+	engine.SetSkillLoader(skillLoader)
+	engine.SetEpisodeContextFunc(episodeCtx)
+	engine.SetUserMessageHandler(userMsgHandler)
+
+	for i := 0; i < 2; i++ {
+		if _, err := engine.Run(context.Background(), "same task twice"); err != nil {
+			t.Fatalf("Run() error: %v", err)
+		}
+	}
+	if skillLoadCount != 2 {
+		t.Errorf("SkillLoader called %d times across 2 identical runs, want 2", skillLoadCount)
+	}
+	if episodeSearchCount != 2 {
+		t.Errorf("episodeCtx called %d times across 2 identical runs, want 2", episodeSearchCount)
+	}
+	if userMsgCount != 2 {
+		t.Errorf("userMsgHandler called %d times across 2 identical runs, want 2", userMsgCount)
+	}
+}
+
 func TestEngine_ToolEventHandler(t *testing.T) {
 	// Verify that ToolEventHandler fires tool_call before and tool_result
 	// after each tool invocation, and does so live (during the loop).

internal/memory/episode_index_http_test.go

@@ -177,3 +177,45 @@ func TestEpisodeIndexRebuildBackoff(t *testing.T) {
 		t.Error("failedAt should be recent")
 	}
 }
+
+// TestSharedEmbedderWarmsDedupAndRebuild verifies the production wiring
+// (NewMemoryManager → embedding.Shared): with an HTTP backend the write-time
+// dedup pass and the vector-index rebuild share ONE cache-warm embedder, so
+// each corpus text is embedded once per process instead of once per pass.
+func TestSharedEmbedderWarmsDedupAndRebuild(t *testing.T) {
+	resetEpIdxes()
+	srv, _, texts := mockEmbedServer(t)
+	dir := t.TempDir()
+
+	cfg := DefaultMemoryConfig()
+	cfg.Embedding = &EmbeddingConfig{Provider: "http", BaseURL: srv.URL + "/v1", Model: "mock-embed"}
+	mm := NewMemoryManager(dir, nil, cfg)
+
+	write := func(id, summary string) {
+		if err := mm.episodes.WriteWithProvenance(id, summary, 5, EpisodeProvenance{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	write("sess-1", "investigated the feline behavior module")
+	if _, err := mm.episodes.recallByVector("cats", 1); err != nil { // triggers the rebuild
+		t.Fatal(err)
+	}
+	afterFirst := texts.Load()
+
+	write("sess-2", "tuned postgres sql indexes")
+	// The dedup pass fits [sess-1, sess-2] — with a shared embedder only the
+	// NEW text is a cache miss. Two cold embedders would re-embed both.
+	if got := texts.Load() - afterFirst; got != 1 {
+		t.Errorf("dedup embedded %d texts for the second write, want 1 (shared cache)", got)
+	}
+
+	afterSecond := texts.Load()
+	if _, err := mm.episodes.recallByVector("database", 1); err != nil { // triggers another rebuild
+		t.Fatal(err)
+	}
+	// The rebuild must be fully cache-warm: only the recall query embeds.
+	if got := texts.Load() - afterSecond; got != 1 {
+		t.Errorf("rebuild + recall embedded %d texts, want 1 (query only)", got)
+	}
+}