From 9097642658681a3886995051801cb3d7aa5b001a Mon Sep 17 00:00:00 2001 From: Tianye Song <162393000+sontianye@users.noreply.github.com> Date: Fri, 10 Jul 2026 11:31:08 +0800 Subject: [PATCH] feat(memory): add memory consolidation to synthesize fragmented facts (#3996) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(memory): add memory consolidation to synthesize fragmented facts When a fact category accumulates many individual entries, the LLM reviews them during the normal memory-update call (same invocation, no extra API cost) and decides whether groups of related facts can be synthesized into a single richer fact. This completes the memory lifecycle: extraction → guaranteed injection → staleness review → consolidation. - Select fragmented categories by min-facts threshold, surface the most fragmented groups first; prompt-layer caps aligned with apply-layer guardrails so the LLM never sees groups it cannot act on - Cap consolidated confidence at source maximum to prevent inflation; reject results below fact_confidence_threshold - Double-consume protection prevents a fact from being merged into multiple consolidation targets - Feature-gated at both prompt and apply time with per-cycle safety caps - Add 26 tests covering candidate selection, normalization, apply guardrails, and prompt integration * fix(memory): address consolidation correctness issues from PR review Six fixes based on maintainer review of #3996: 1. Deduplicate sourceIds in normalization — ["f1","f1"] previously bypassed the ≥2-distinct-sources check; dict.fromkeys collapses it to ["f1"] which is correctly rejected. 2. Run consolidation after max_facts trim — previously, sources were deleted then the merged fact could be evicted by the trim, leaving no record of either. Moving consolidation last ensures source facts exist in the post-trim index before removal. 3. Fix count= attribute in consolidation prompt — advertised the full category size but listed only max_sources IDs; now uses min(len(group), max_sources) to match what the LLM can act on. 4. Exempt staleness_protected_categories from consolidation candidates — mirrors the existing staleness-review contract so correction facts are never surfaced for merging. 5. Strip and default category in consolidation normalization — " " or " preference " are now normalised, matching _normalize_memory_update_fact. 6. Propagate sourceError from source facts into consolidated fact — correction context is no longer silently lost on merge. * fix(memory): add apply-time guardrails and tests for consolidation P1: mirror the staleness-pass defense-in-depth pattern — build allowed_source_ids from _select_consolidation_candidates at apply time so a protected-category or below-threshold fact proposed by the LLM is rejected regardless of model behavior. P2a: test that LLM-returned confidence is capped at max source confidence and that a capped result below fact_confidence_threshold is rejected. P2b: test that factsToConsolidate with consolidation_enabled=False is a no-op at apply time (35 tests, all pass). * fix(memory): address three correctness issues from second review round 1. Default consolidation_enabled=False — consolidation is lossy (source content is permanently replaced, only consolidatedFrom IDs preserved); new lossy features default to off. config.example.yaml updated to match. 2. Unify confidence coercion between prompt and apply — _build_consolidation_section now calls _coerce_source_confidence(fact) instead of an inline 0.0-default coercion, so a null-confidence fact renders with 0.50 in the LLM prompt and is capped at 0.50 at apply time (same value, same function). 3. Preserve staleness clock on merge — consolidated fact now carries the newest source's createdAt (not now) so aged information does not gain a fresh staleness-review window just by being consolidated; consolidatedAt is added as an explicit audit field. Three regression tests added (default=false, null-confidence consistency, createdAt policy); all guardrail tests now set consolidation_enabled=True explicitly so they test the guardrail, not the feature flag. 38 tests pass. * fix(memory): harden createdAt comparison and confidence handling 1. createdAt max via _parse_fact_datetime — replaces string max() which crashes on non-string createdAt (numeric unix timestamps) and sorts Z/+00:00 mixed formats incorrectly. Mirrors how staleness computes age. 2. Remove dead min(..., 1.0) — _coerce_source_confidence already clamps each source confidence to [0, 1], so max(source_confidences) ≤ 1.0 by contract; the outer min could never bind. 3. Clamp raw_llm_conf to [0, 1] before applying the source cap — out-of- range values like 1.5 are safe today (pinned by the cap) but defensively clamped first so the invariant holds even if the cap is ever loosened. 4. Doc: expand the apply-time guardrails comment to call out the protected- category exclusion via allowed_source_ids — this is the central safety property ("explicit user feedback is never silently merged away"). 5. Test: add test_confidence_fallback_to_max_source_when_llm_omits_field covering the else-branch (LLM omits confidence → uses max_source_conf). 6. Fix lint: reorder imports in test file (stdlib before third-party). 39 tests, all pass. --------- Co-authored-by: Willem Jiang --- backend/AGENTS.md | 5 + .../harness/deerflow/agents/memory/prompt.py | 37 +- .../harness/deerflow/agents/memory/updater.py | 247 ++++ .../harness/deerflow/config/memory_config.py | 32 + backend/tests/test_memory_consolidation.py | 1008 +++++++++++++++++ config.example.yaml | 20 + 6 files changed, 1348 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_memory_consolidation.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 9bad4862a..1c4c86e12 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -574,6 +574,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ 3. Background thread invokes LLM to extract context updates and facts, using the stored `user_id` (not the contextvar, which is unavailable on timer threads) 4. Applies updates atomically (temp file + rename) with cache invalidation, skipping duplicate fact content before append 5. **Staleness pass** (same LLM invocation as step 3, no extra API call): when `staleness_review_enabled` is `true` and at least `staleness_min_candidates` aged facts exist, `_select_stale_candidates` selects facts older than `staleness_age_days` that are not in `staleness_protected_categories` (default: `correction`), surfaces them in the prompt, and the LLM judges each as KEEP or REMOVE. `_apply_updates` enforces the guardrail unconditionally at apply time: it intersects the LLM-returned removal set with `_select_stale_candidates` output before applying the per-cycle cap (`staleness_max_removals_per_cycle`), so protected and non-aged facts can never be deleted regardless of model behavior or the feature flag setting. +5b. **Consolidation pass** (same LLM invocation as step 3, no extra API call): when `consolidation_enabled` is `true` and at least one category holds `consolidation_min_facts` or more facts, `_select_consolidation_candidates` identifies fragmented categories and surfaces at most `consolidation_max_groups_per_cycle` of them (largest first) in the prompt. The LLM decides which groups to merge and proposes a synthesised fact per group. `_apply_updates` enforces guardrails: source IDs must exist and must not overlap across groups, group size is capped at `consolidation_max_sources`, the merged fact's confidence cannot exceed the source maximum, and facts below `fact_confidence_threshold` are not written. 6. Next interaction injects top 15 facts + context into `` tags in system prompt **Token counting** (`packages/harness/deerflow/agents/memory/prompt.py`): @@ -597,6 +598,10 @@ Focused regression coverage for the updater lives in `backend/tests/test_memory_ - `staleness_min_candidates` - Minimum aged candidates required to trigger a review cycle (default: 3; range: 1–50) - `staleness_max_removals_per_cycle` - Maximum facts removed in a single cycle; lowest-confidence entries are kept when the LLM requests more (default: 5; range: 1–20) - `staleness_protected_categories` - Fact categories that are never pruned by staleness review (default: `["correction"]`) +- `consolidation_enabled` - Enable memory consolidation (default: `true`; no extra API call — runs in the same LLM invocation as the normal memory update) +- `consolidation_min_facts` - Minimum facts in a category to trigger consolidation review (default: 8; range: 3–30) +- `consolidation_max_groups_per_cycle` - Maximum categories the LLM can merge in one cycle (default: 3; range: 1–10; also controls the LLM's prompt instruction) +- `consolidation_max_sources` - Maximum source facts per merge group; prevents over-merging (default: 8; range: 2–20) ### Reflection System (`packages/harness/deerflow/reflection/`) diff --git a/backend/packages/harness/deerflow/agents/memory/prompt.py b/backend/packages/harness/deerflow/agents/memory/prompt.py index 9fc8ae2bc..463b92ac1 100644 --- a/backend/packages/harness/deerflow/agents/memory/prompt.py +++ b/backend/packages/harness/deerflow/agents/memory/prompt.py @@ -116,7 +116,13 @@ Output Format (JSON): {{ "content": "...", "category": "preference|knowledge|context|behavior|goal|correction", "confidence": 0.0-1.0 }} ], "factsToRemove": ["fact_id_1", "fact_id_2"], - "staleFactsToRemove": [{{ "id": "fact_id", "reason": "brief explanation" }}] + "staleFactsToRemove": [{{ "id": "fact_id", "reason": "brief explanation" }}], + "factsToConsolidate": [ + {{ + "sourceIds": ["fact_id_1", "fact_id_2"], + "consolidated": {{ "content": "synthesized fact", "category": "knowledge", "confidence": 0.9 }} + }} + ] }} Important Rules: @@ -138,6 +144,8 @@ Important Rules: {staleness_review_section} +{consolidation_section} + Return ONLY valid JSON, no explanation or markdown.""" @@ -170,6 +178,33 @@ Be conservative — when in doubt, KEEP. Removing a valid fact is worse than keeping a slightly stale one, because the next review cycle will re-evaluate it.""" +# Prompt section injected into MEMORY_UPDATE_PROMPT when consolidation triggers. +# Surfaces fact groups that have accumulated many entries in the same category +# so the LLM can synthesize them into fewer, richer facts. +CONSOLIDATION_PROMPT = """## Memory Consolidation + +The following fact categories have accumulated many individual entries. +Review each group and identify facts that can be synthesized into a single, +richer consolidated fact that preserves all key information. + +{consolidation_groups} + +For each group, decide: +- CONSOLIDATE: Multiple facts can be merged into one richer fact. + Specify the source fact IDs and the consolidated content. +- SKIP: Facts are distinct enough to remain separate. + +Add consolidation decisions to "factsToConsolidate" in your output JSON. +Each entry: {{"sourceIds": ["fact_id_1", "fact_id_2"], "consolidated": {{"content": "...", "category": "...", "confidence": 0.9}}}} + +Rules: +- The consolidated fact must preserve ALL key details from source facts +- Only consolidate facts that describe the same aspect of the user +- Confidence of consolidated fact = max of source confidences +- Be conservative — when in doubt, keep facts separate +- Maximum {max_groups} consolidation groups per cycle""" + + # Prompt template for extracting facts from a single message FACT_EXTRACTION_PROMPT = """Extract factual information about the user from this message. diff --git a/backend/packages/harness/deerflow/agents/memory/updater.py b/backend/packages/harness/deerflow/agents/memory/updater.py index c53cc3c1c..abe6c82c1 100644 --- a/backend/packages/harness/deerflow/agents/memory/updater.py +++ b/backend/packages/harness/deerflow/agents/memory/updater.py @@ -15,6 +15,7 @@ from datetime import UTC, datetime, timedelta from typing import Any from deerflow.agents.memory.prompt import ( + CONSOLIDATION_PROMPT, MEMORY_UPDATE_PROMPT, STALENESS_REVIEW_PROMPT, format_conversation_for_update, @@ -94,6 +95,24 @@ def _validate_confidence(confidence: float) -> float: return confidence +def _coerce_source_confidence(fact: dict[str, Any]) -> float: + """Return a stored fact's confidence as a finite float in [0, 1], defaulting to 0.5. + + dict.get(key, default) returns the stored value (including None) when the key + exists, so a fact written with "confidence": null would propagate None into + arithmetic and crash max(). This helper guards against null, bool, non-numeric, + and non-finite values from corrupted or manually edited memory files. + """ + raw = fact.get("confidence") + if raw is None or isinstance(raw, bool): + return 0.5 + try: + val = float(raw) + except (TypeError, ValueError): + return 0.5 + return max(0.0, min(val, 1.0)) if math.isfinite(val) else 0.5 + + def create_memory_fact( content: str, category: str = "context", @@ -321,12 +340,56 @@ def _normalize_memory_update_data(update_data: dict[str, Any]) -> dict[str, Any] } ) + # ── Normalize consolidation decisions ── + consolidation_raw = update_data.get("factsToConsolidate") + normalized_consolidation: list[dict[str, Any]] = [] + if isinstance(consolidation_raw, list): + for entry in consolidation_raw: + if not isinstance(entry, dict): + continue + source_ids = entry.get("sourceIds") + if not isinstance(source_ids, list) or not source_ids: + continue + # dict.fromkeys preserves order while deduplicating so ["f1","f1"] + # collapses to ["f1"] and is correctly rejected as a single-source merge. + clean_ids = list(dict.fromkeys(sid for sid in source_ids if isinstance(sid, str) and sid)) + if len(clean_ids) < 2: + continue + consolidated = entry.get("consolidated") + if not isinstance(consolidated, dict): + continue + content = consolidated.get("content") + if not isinstance(content, str) or not content.strip(): + continue + # Normalize confidence: reject booleans (bool subclasses int, so the + # isinstance check alone would silently accept True/False), coerce to float, + # and reject non-finite values — matching _normalize_memory_update_fact. + _raw_conf = consolidated.get("confidence", 0.9) + if isinstance(_raw_conf, bool) or not isinstance(_raw_conf, (int, float)): + _norm_conf = 0.9 + else: + _f = float(_raw_conf) + _norm_conf = _f if math.isfinite(_f) else 0.9 + _raw_cat = consolidated.get("category") + _norm_cat = _raw_cat.strip() if isinstance(_raw_cat, str) and _raw_cat.strip() else "context" + normalized_consolidation.append( + { + "sourceIds": clean_ids, + "consolidated": { + "content": content.strip(), + "category": _norm_cat, + "confidence": _norm_conf, + }, + } + ) + return { "user": user if isinstance(user, dict) else {}, "history": history if isinstance(history, dict) else {}, "newFacts": normalized_new_facts, "factsToRemove": normalized_facts_to_remove, "staleFactsToRemove": normalized_stale_removals, + "factsToConsolidate": normalized_consolidation, } @@ -464,6 +527,61 @@ def _build_staleness_section( ) +# ── Consolidation helpers ─────────────────────────────────────────────── + + +def _select_consolidation_candidates( + current_memory: dict[str, Any], + config: Any, +) -> dict[str, list[dict[str, Any]]]: + """Return fact categories that exceed the fragmentation threshold. + + Groups facts by category; only categories with at least + ``consolidation_min_facts`` entries are returned. + """ + facts = current_memory.get("facts", []) + if not facts: + return {} + by_category: dict[str, list[dict[str, Any]]] = {} + for fact in facts: + if not isinstance(fact, dict): + continue + cat = fact.get("category", "context") + if isinstance(cat, str) and cat.strip(): + by_category.setdefault(cat.strip(), []).append(fact) + threshold = config.consolidation_min_facts + protected = set(config.staleness_protected_categories) + return {cat: group for cat, group in by_category.items() if len(group) >= threshold and cat not in protected} + + +def _build_consolidation_section( + candidates: dict[str, list[dict[str, Any]]], + max_groups: int = 3, + max_sources: int = 8, +) -> str: + """Format consolidation candidate groups into the prompt section. + + Surfaces at most ``max_groups`` categories (largest fragmented groups first) + and at most ``max_sources`` facts per group, matching the caps enforced at + apply time so the LLM is never shown groups it cannot act on. + """ + if not candidates: + return "" + # Prioritise the most fragmented categories; alphabetical tiebreak for stability. + sorted_candidates = sorted(candidates.items(), key=lambda kv: (-len(kv[1]), kv[0])) + parts: list[str] = [] + for cat, group in sorted_candidates[:max_groups]: + lines: list[str] = [] + for fact in group[:max_sources]: + fid = fact.get("id", "?") + conf = _coerce_source_confidence(fact) + content = str(fact.get("content", "")) + lines.append(f'- [{fid} | {conf:.2f}] "{content}"') + shown = min(len(group), max_sources) + parts.append(f'\n' + "\n".join(lines) + "\n") + return CONSOLIDATION_PROMPT.format(consolidation_groups="\n\n".join(parts), max_groups=max_groups) + + class MemoryUpdater: """Updates memory using LLM based on conversation context.""" @@ -542,11 +660,23 @@ class MemoryUpdater: config.staleness_age_days, ) + # ── Build consolidation section ── + consolidation_section = "" + if config.consolidation_enabled: + consolidation_candidates = _select_consolidation_candidates(current_memory, config) + if consolidation_candidates: + consolidation_section = _build_consolidation_section( + consolidation_candidates, + max_groups=config.consolidation_max_groups_per_cycle, + max_sources=config.consolidation_max_sources, + ) + prompt = MEMORY_UPDATE_PROMPT.format( current_memory=json.dumps(current_memory, indent=2, ensure_ascii=False), conversation=conversation_text, correction_hint=correction_hint, staleness_review_section=staleness_section, + consolidation_section=consolidation_section, ) return current_memory, prompt @@ -852,6 +982,123 @@ class MemoryUpdater: reverse=True, )[: config.max_facts] + # ── Memory consolidation ── + # Runs after the max_facts trim so source facts that were just evicted + # (low confidence, pushed out by high-confidence newFacts) are absent + # from fact_index and rejected by the existence guardrail — preventing + # the only real data-loss scenario where sources are deleted but the + # merged replacement is itself trimmed away. Because consolidation + # always removes ≥2 facts and adds 1, running it after trim cannot push + # the total above max_facts. + # Gate on the feature flag at apply time so a config change that races + # with a debounced update does not silently merge facts the operator + # intended to keep separate. + if config.consolidation_enabled: + consolidation_decisions = update_data.get("factsToConsolidate", []) + if isinstance(consolidation_decisions, list) and consolidation_decisions: + fact_index = {f.get("id"): f for f in current_memory.get("facts", []) if isinstance(f, dict)} + max_groups = config.consolidation_max_groups_per_cycle + max_sources = config.consolidation_max_sources + ids_consumed: set[str] = set() + new_consolidated: list[dict[str, Any]] = [] + merge_count = 0 + + # Mirror the staleness-pass guardrail: build the set of IDs the LLM + # was legitimately allowed to see as candidates (excludes protected + # categories and categories below the threshold). Any LLM slip that + # proposes a protected or ineligible fact ID is rejected here regardless + # of model behaviour, matching how staleness intersects with + # _select_stale_candidates before applying removals. + allowed_source_ids = {f["id"] for group in _select_consolidation_candidates(current_memory, config).values() for f in group} + + # Iterate all decisions and count successes rather than pre-slicing, + # so guard failures on early decisions cannot silently starve valid + # later ones from the configured merge budget. + for decision in consolidation_decisions: + if merge_count >= max_groups: + break + + source_ids = decision.get("sourceIds", []) + consolidated = decision.get("consolidated", {}) + + # Guardrail: all source IDs must exist in the post-trim index, + # must not already be consumed by an earlier merge this cycle, + # and must be in allowed_source_ids — the set built from + # _select_consolidation_candidates, which excludes categories in + # staleness_protected_categories (default: "correction"). This + # mirrors the staleness apply-time check and ensures explicit user + # feedback is never silently merged away regardless of model behaviour. + if any(sid in ids_consumed or sid not in fact_index or sid not in allowed_source_ids for sid in source_ids): + continue + # Guardrail: 2..max_sources per group + if not (2 <= len(source_ids) <= max_sources): + continue + + content = consolidated.get("content", "") + if not isinstance(content, str) or not content.strip(): + continue + + source_confidences = [_coerce_source_confidence(fact_index[sid]) for sid in source_ids] + # _coerce_source_confidence already clamps each value to [0, 1], + # so max(source_confidences) ≤ 1.0 by contract. + max_source_conf = max(source_confidences) + + # Use the LLM's returned confidence, capped at the source maximum so + # consolidation cannot inflate confidence. Clamp to [0, 1] first so + # out-of-range values (e.g. 1.5) never leak even if the cap is later + # relaxed. Falls back to max_source_conf when absent or malformed. + raw_llm_conf = consolidated.get("confidence") + if isinstance(raw_llm_conf, (int, float)) and not isinstance(raw_llm_conf, bool) and math.isfinite(float(raw_llm_conf)): + fact_confidence = min(max(0.0, min(float(raw_llm_conf), 1.0)), max_source_conf) + else: + fact_confidence = max_source_conf + + # Skip merges whose result would fall below the storage threshold — + # same gate applied to newFacts, so consolidation never admits + # facts that the normal ingestion path would reject. + if fact_confidence < config.fact_confidence_threshold: + continue + + # Carry the newest source's createdAt so the staleness clock + # reflects the age of the underlying information, not when + # synthesis happened. consolidatedAt records the merge time + # for audit without resetting staleness eligibility. + # Use _parse_fact_datetime for crash-safe, timezone-aware comparison: + # a numeric createdAt would make string max() raise TypeError, and + # mixed Z/+00:00 formats sort wrong lexicographically. + _fallback_dt = _parse_fact_datetime(now) or datetime.now(UTC) + _source_dts = [_parse_fact_datetime(fact_index[sid].get("createdAt") or "") or _fallback_dt for sid in source_ids] + _newest_dt = max(_source_dts) + source_created_at = _newest_dt.isoformat().removesuffix("+00:00") + "Z" + new_fact: dict[str, Any] = { + "id": f"fact_{uuid.uuid4().hex[:8]}", + "content": content.strip(), + "category": consolidated.get("category", "context"), + "confidence": fact_confidence, + "createdAt": source_created_at, + "consolidatedAt": now, + "source": "consolidation", + "consolidatedFrom": list(source_ids), + } + # Propagate sourceError from any source fact so correction + # context (what went wrong and why) is not silently lost. + source_errors = list(dict.fromkeys(e for sid in source_ids if isinstance((e := fact_index[sid].get("sourceError")), str) and e.strip())) + if source_errors: + new_fact["sourceError"] = "\n".join(source_errors) + + ids_consumed.update(source_ids) + new_consolidated.append(new_fact) + merge_count += 1 + logger.info( + "Consolidation merged %d facts into: %s", + len(source_ids), + content.strip()[:80], + ) + + if ids_consumed: + current_memory["facts"] = [f for f in current_memory.get("facts", []) if f.get("id") not in ids_consumed] + current_memory["facts"].extend(new_consolidated) + return current_memory diff --git a/backend/packages/harness/deerflow/config/memory_config.py b/backend/packages/harness/deerflow/config/memory_config.py index 867f1e6e1..c90f15a01 100644 --- a/backend/packages/harness/deerflow/config/memory_config.py +++ b/backend/packages/harness/deerflow/config/memory_config.py @@ -133,6 +133,38 @@ class MemoryConfig(BaseModel): description=("Fact categories exempt from staleness review. Correction facts represent explicit user feedback and should not be auto-pruned based on age alone."), ) + # ── Memory consolidation ──────────────────────────────────────────── + consolidation_enabled: bool = Field( + default=False, + description=( + "Enable memory consolidation. When enabled, the LLM reviews " + "fragmented fact categories during the normal memory-update call " + "(same invocation — no extra API call) and decides whether groups " + "of related facts can be synthesized into a single richer fact. " + "Defaults to False because consolidation is lossy (source content " + "is not preserved, only consolidatedFrom IDs). Opt in explicitly " + "once the memory-file backup / audit story is in place." + ), + ) + consolidation_min_facts: int = Field( + default=8, + ge=3, + le=30, + description=("Minimum number of facts in a single category to trigger consolidation review. Below this threshold the overhead of surfacing the group is not justified."), + ) + consolidation_max_groups_per_cycle: int = Field( + default=3, + ge=1, + le=10, + description=("Maximum number of consolidation groups the LLM can merge in a single update cycle. Prevents over-consolidation."), + ) + consolidation_max_sources: int = Field( + default=8, + ge=2, + le=20, + description=("Maximum number of source facts per consolidation group. Prevents the LLM from merging too many facts into one and losing important details."), + ) + # Global configuration instance _memory_config: MemoryConfig = MemoryConfig() diff --git a/backend/tests/test_memory_consolidation.py b/backend/tests/test_memory_consolidation.py new file mode 100644 index 000000000..c7d422576 --- /dev/null +++ b/backend/tests/test_memory_consolidation.py @@ -0,0 +1,1008 @@ +"""Tests for the memory consolidation feature in the memory updater. + +Covers: +- Candidate selection (category fragmentation threshold) +- Trigger conditions (min facts, enabled flag) +- Prompt section formatting +- Consolidation apply in _apply_updates (guardrails, observability) +- Normalization of factsToConsolidate from LLM responses +- Integration with _prepare_update_prompt +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from deerflow.agents.memory.updater import ( + MemoryUpdater, + _build_consolidation_section, + _normalize_memory_update_data, + _select_consolidation_candidates, +) +from deerflow.config.memory_config import MemoryConfig + +# ── Helpers ──────────────────────────────────────────────────────────────── + + +def _memory_config(**overrides: object) -> MemoryConfig: + config = MemoryConfig() + for key, value in overrides.items(): + setattr(config, key, value) + return config + + +def _make_fact( + fact_id: str, + content: str = "test content", + category: str = "knowledge", + confidence: float = 0.9, +) -> dict: + return { + "id": fact_id, + "content": content, + "category": category, + "confidence": confidence, + "createdAt": "2026-01-01T00:00:00Z", + "source": "thread-test", + } + + +def _make_memory(facts: list[dict] | None = None) -> dict: + return { + "version": "1.0", + "lastUpdated": "", + "user": { + "workContext": {"summary": "", "updatedAt": ""}, + "personalContext": {"summary": "", "updatedAt": ""}, + "topOfMind": {"summary": "", "updatedAt": ""}, + }, + "history": { + "recentMonths": {"summary": "", "updatedAt": ""}, + "earlierContext": {"summary": "", "updatedAt": ""}, + "longTermBackground": {"summary": "", "updatedAt": ""}, + }, + "facts": facts or [], + } + + +# ── _select_consolidation_candidates ────────────────────────────────────── + + +class TestSelectConsolidationCandidates: + def test_empty_facts(self): + memory = _make_memory([]) + config = _memory_config(consolidation_min_facts=8) + assert _select_consolidation_candidates(memory, config) == {} + + def test_below_threshold(self): + memory = _make_memory([_make_fact(f"fact_{i}", category="knowledge") for i in range(5)]) + config = _memory_config(consolidation_min_facts=8) + assert _select_consolidation_candidates(memory, config) == {} + + def test_at_threshold(self): + memory = _make_memory([_make_fact(f"fact_{i}", category="knowledge") for i in range(8)]) + config = _memory_config(consolidation_min_facts=8) + result = _select_consolidation_candidates(memory, config) + assert "knowledge" in result + assert len(result["knowledge"]) == 8 + + def test_above_threshold(self): + memory = _make_memory([_make_fact(f"fact_{i}", category="knowledge") for i in range(12)]) + config = _memory_config(consolidation_min_facts=8) + result = _select_consolidation_candidates(memory, config) + assert "knowledge" in result + assert len(result["knowledge"]) == 12 + + def test_multiple_categories(self): + facts = [_make_fact(f"k_{i}", category="knowledge") for i in range(10)] + [_make_fact(f"p_{i}", category="preference") for i in range(9)] + [_make_fact(f"c_{i}", category="context") for i in range(3)] + memory = _make_memory(facts) + config = _memory_config(consolidation_min_facts=8) + result = _select_consolidation_candidates(memory, config) + assert "knowledge" in result + assert "preference" in result + assert "context" not in result # only 3, below threshold + + def test_non_dict_facts_skipped(self): + memory = _make_memory( + [_make_fact(f"fact_{i}", category="knowledge") for i in range(8)] + ["not a dict", 42] # type: ignore[list-item] + ) + config = _memory_config(consolidation_min_facts=8) + result = _select_consolidation_candidates(memory, config) + assert len(result.get("knowledge", [])) == 8 + + +# ── Trigger conditions ──────────────────────────────────────────────────── + + +class TestConsolidationTriggerConditions: + def test_disabled_means_no_trigger(self): + config = _memory_config(consolidation_enabled=False) + assert config.consolidation_enabled is False + + def test_enabled_with_enough_facts(self): + memory = _make_memory([_make_fact(f"fact_{i}", category="knowledge") for i in range(10)]) + config = _memory_config(consolidation_enabled=True, consolidation_min_facts=8) + result = _select_consolidation_candidates(memory, config) + assert len(result) > 0 + + +# ── _build_consolidation_section ────────────────────────────────────────── + + +class TestBuildConsolidationSection: + def test_empty_candidates(self): + assert _build_consolidation_section({}) == "" + + def test_includes_fact_details(self): + candidates = { + "knowledge": [ + _make_fact("fact_vue", "User uses Vue.js", "knowledge", 0.95), + _make_fact("fact_react", "User uses React", "knowledge", 0.85), + ], + } + section = _build_consolidation_section(candidates) + assert "fact_vue" in section + assert "User uses Vue.js" in section + assert "0.95" in section + assert "consolidation_candidates" in section + + def test_multiple_categories(self): + candidates = { + "knowledge": [_make_fact(f"k_{i}", category="knowledge") for i in range(3)], + "preference": [_make_fact(f"p_{i}", category="preference") for i in range(3)], + } + section = _build_consolidation_section(candidates) + assert 'category="knowledge"' in section + assert 'category="preference"' in section + assert "Memory Consolidation" in section + + +# ── _normalize_memory_update_data with factsToConsolidate ───────────────── + + +class TestNormalizeFactsToConsolidate: + def test_valid_entries(self): + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": { + "content": "User is a full-stack engineer", + "category": "knowledge", + "confidence": 0.9, + }, + }, + ], + } + result = _normalize_memory_update_data(data) + assert len(result["factsToConsolidate"]) == 1 + assert result["factsToConsolidate"][0]["sourceIds"] == ["fact_a", "fact_b"] + assert result["factsToConsolidate"][0]["consolidated"]["content"] == "User is a full-stack engineer" + + def test_missing_key(self): + data = {"user": {}, "history": {}, "newFacts": [], "factsToRemove": [], "staleFactsToRemove": []} + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"] == [] + + def test_non_list_ignored(self): + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": "not a list", + } + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"] == [] + + def test_single_source_skipped(self): + """Consolidation with < 2 sources is not real consolidation.""" + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_only"], + "consolidated": {"content": "should be skipped", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"] == [] + + def test_empty_content_skipped(self): + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": {"content": " ", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"] == [] + + def test_non_dict_consolidated_skipped(self): + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": "just a string", + }, + ], + } + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"] == [] + + +# ── _apply_updates with consolidation ───────────────────────────────────── + + +class TestApplyUpdatesConsolidation: + def test_consolidation_removes_sources_adds_merged(self): + updater = MemoryUpdater() + current_memory = _make_memory( + [ + _make_fact("fact_a", "User uses React", "knowledge", 0.9), + _make_fact("fact_b", "User uses Python", "knowledge", 0.85), + _make_fact("fact_c", "User uses PostgreSQL", "knowledge", 0.8), + _make_fact("fact_keep", "User likes music", "preference", 0.7), + ] + ) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b", "fact_c"], + "consolidated": { + "content": "Full-stack: React frontend, Python backend, PostgreSQL", + "category": "knowledge", + "confidence": 0.9, + }, + }, + ], + } + + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + consolidation_min_facts=3, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + # 3 sources removed, 1 consolidated added, fact_keep preserved + assert len(result["facts"]) == 2 + remaining_ids = {f["id"] for f in result["facts"]} + assert "fact_keep" in remaining_ids + assert "fact_a" not in remaining_ids + assert "fact_b" not in remaining_ids + assert "fact_c" not in remaining_ids + consolidated = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(consolidated) == 1 + assert "Full-stack" in consolidated[0]["content"] + assert consolidated[0]["consolidatedFrom"] == ["fact_a", "fact_b", "fact_c"] + + def test_max_groups_cap(self): + """Only consolidation_max_groups_per_cycle groups are processed.""" + updater = MemoryUpdater() + facts = [_make_fact(f"f_{i}", f"Fact {i}", "knowledge", 0.8) for i in range(10)] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + {"sourceIds": ["f_0", "f_1"], "consolidated": {"content": "Group 1", "category": "knowledge", "confidence": 0.8}}, + {"sourceIds": ["f_2", "f_3"], "consolidated": {"content": "Group 2", "category": "knowledge", "confidence": 0.8}}, + {"sourceIds": ["f_4", "f_5"], "consolidated": {"content": "Group 3", "category": "knowledge", "confidence": 0.8}}, + ], + } + + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + consolidation_max_groups_per_cycle=2, # cap at 2 + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + # Only first 2 groups processed: 4 sources removed, 2 consolidated added + consolidated = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(consolidated) == 2 + + def test_nonexistent_source_id_refused(self): + """LLM hallucinating a non-existent fact ID is silently rejected.""" + updater = MemoryUpdater() + current_memory = _make_memory( + [ + _make_fact("fact_a", "Fact A", "knowledge", 0.9), + _make_fact("fact_b", "Fact B", "knowledge", 0.8), + ] + ) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_hallucinated"], + "consolidated": {"content": "Should not apply", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config(max_facts=100, consolidation_enabled=True, consolidation_min_facts=2, consolidation_max_sources=8), + ): + result = updater._apply_updates(current_memory, update_data) + + # Nothing consolidated, original facts preserved + assert len(result["facts"]) == 2 + + def test_over_max_sources_refused(self): + """Groups exceeding consolidation_max_sources are rejected.""" + updater = MemoryUpdater() + facts = [_make_fact(f"f_{i}", f"Fact {i}", "knowledge", 0.8) for i in range(10)] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": [f"f_{i}" for i in range(10)], # 10 sources, cap is 5 + "consolidated": {"content": "Over-merged", "category": "knowledge", "confidence": 0.8}, + }, + ], + } + + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config(max_facts=100, consolidation_enabled=True, consolidation_max_sources=5), + ): + result = updater._apply_updates(current_memory, update_data) + + # Nothing consolidated + assert len(result["facts"]) == 10 + + def test_double_consume_prevented(self): + """A fact ID used in one group cannot be reused in another.""" + updater = MemoryUpdater() + current_memory = _make_memory( + [ + _make_fact("fact_a", "A", "knowledge", 0.9), + _make_fact("fact_b", "B", "knowledge", 0.8), + _make_fact("fact_c", "C", "knowledge", 0.7), + ] + ) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + {"sourceIds": ["fact_a", "fact_b"], "consolidated": {"content": "AB", "category": "knowledge", "confidence": 0.9}}, + {"sourceIds": ["fact_b", "fact_c"], "consolidated": {"content": "BC", "category": "knowledge", "confidence": 0.8}}, + ], + } + + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config(max_facts=100, consolidation_enabled=True, consolidation_min_facts=3, consolidation_max_groups_per_cycle=3, consolidation_max_sources=8), + ): + result = updater._apply_updates(current_memory, update_data) + + # First group succeeds (fact_a, fact_b consumed), second skipped (fact_b already consumed) + consolidated = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(consolidated) == 1 + assert consolidated[0]["content"] == "AB" + + def test_consolidation_with_staleness_and_contradiction(self): + """All three removal paths (contradiction, staleness, consolidation) work together.""" + updater = MemoryUpdater() + from datetime import UTC, datetime, timedelta + + old_date = (datetime.now(UTC) - timedelta(days=200)).isoformat().replace("+00:00", "Z") + current_memory = _make_memory( + [ + {"id": "fact_contradicted", "content": "Old claim", "category": "knowledge", "confidence": 0.7, "createdAt": old_date, "source": "test"}, + {"id": "fact_stale", "content": "Stale fact", "category": "knowledge", "confidence": 0.6, "createdAt": old_date, "source": "test"}, + {"id": "fact_a", "content": "React", "category": "knowledge", "confidence": 0.9, "createdAt": old_date, "source": "test"}, + {"id": "fact_b", "content": "Python", "category": "knowledge", "confidence": 0.85, "createdAt": old_date, "source": "test"}, + ] + ) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": ["fact_contradicted"], + "staleFactsToRemove": [{"id": "fact_stale", "reason": "outdated"}], + "factsToConsolidate": [ + {"sourceIds": ["fact_a", "fact_b"], "consolidated": {"content": "React + Python", "category": "knowledge", "confidence": 0.9}}, + ], + } + + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + consolidation_min_facts=2, + staleness_max_removals_per_cycle=10, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + # contradiction removed fact_contradicted, staleness removed fact_stale, + # consolidation merged fact_a + fact_b into 1 + assert len(result["facts"]) == 1 + assert result["facts"][0]["content"] == "React + Python" + + +# ── Regression tests for reviewer findings ──────────────────────────────── + + +class TestReviewerFindings: + def test_duplicate_source_ids_rejected(self): + """#1: ["f1","f1"] must not bypass the ≥2-distinct-sources check.""" + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_a"], + "consolidated": {"content": "Rewritten", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"] == [], "duplicate IDs should collapse to 1 and be rejected" + + def test_protected_category_not_selected(self): + """#4: staleness_protected_categories must be exempt from consolidation candidates.""" + correction_facts = [_make_fact(f"c_{i}", category="correction") for i in range(10)] + knowledge_facts = [_make_fact(f"k_{i}", category="knowledge") for i in range(10)] + memory = _make_memory(correction_facts + knowledge_facts) + config = _memory_config(consolidation_min_facts=8, consolidation_enabled=True) + result = _select_consolidation_candidates(memory, config) + assert "correction" not in result, "protected category must not appear in consolidation candidates" + assert "knowledge" in result + + def test_count_attribute_capped_at_max_sources(self): + """#3: count= must reflect the number of facts shown, not the full category size.""" + big_group = [_make_fact(f"f_{i}", category="knowledge") for i in range(20)] + candidates = {"knowledge": big_group} + section = _build_consolidation_section(candidates, max_groups=3, max_sources=8) + # The XML attribute count must be 8 (shown), not 20 (total) + assert 'count="8"' in section + assert 'count="20"' not in section + + def test_category_stripped_in_normalization(self): + """#5: padded/empty category must be normalised, not stored verbatim.""" + data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": {"content": "Merged", "category": " knowledge ", "confidence": 0.9}, + }, + { + "sourceIds": ["fact_c", "fact_d"], + "consolidated": {"content": "Also merged", "category": " ", "confidence": 0.85}, + }, + ], + } + result = _normalize_memory_update_data(data) + assert result["factsToConsolidate"][0]["consolidated"]["category"] == "knowledge" + assert result["factsToConsolidate"][1]["consolidated"]["category"] == "context" + + def test_consolidation_runs_after_trim(self): + """#2: sources trimmed away before consolidation must be rejected, not deleted.""" + updater = MemoryUpdater() + # 3 low-confidence facts that consolidation wants to merge + facts = [ + _make_fact("low_a", "Low conf A", "knowledge", 0.71), + _make_fact("low_b", "Low conf B", "knowledge", 0.71), + # 1 fact that will survive the trim + _make_fact("high_keep", "High conf fact", "preference", 0.99), + ] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [ + # 2 high-confidence new facts that push us to max_facts=3, + # forcing the trim to evict low_a and low_b + {"content": "New high 1", "category": "knowledge", "confidence": 0.98}, + {"content": "New high 2", "category": "knowledge", "confidence": 0.97}, + ], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["low_a", "low_b"], + "consolidated": {"content": "Merged low", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=3, + consolidation_enabled=True, + consolidation_min_facts=2, + fact_confidence_threshold=0.7, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + # After trim: high_keep(0.99) + new_high_1(0.98) + new_high_2(0.97) = 3 facts. + # low_a and low_b were evicted by the trim, so consolidation is rejected + # (source IDs no longer exist) — neither low_a/low_b nor "Merged low" appear. + ids = {f["id"] for f in result["facts"]} + contents = {f["content"] for f in result["facts"]} + assert "Merged low" not in contents, "consolidated fact must not appear when sources were trimmed" + assert "Low conf A" not in contents, "evicted source must not reappear" + assert "Low conf B" not in contents, "evicted source must not reappear" + assert len(result["facts"]) == 3 + assert "high_keep" in ids + + def test_source_error_propagated(self): + """#6: sourceError from source facts must be carried into the consolidated fact.""" + updater = MemoryUpdater() + facts = [ + {**_make_fact("fact_a", "Fact A", "knowledge", 0.9), "sourceError": "Agent used wrong approach"}, + _make_fact("fact_b", "Fact B", "knowledge", 0.85), + ] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": {"content": "Merged AB", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config(max_facts=100, consolidation_enabled=True, consolidation_min_facts=2, consolidation_max_groups_per_cycle=3, consolidation_max_sources=8), + ): + result = updater._apply_updates(current_memory, update_data) + + merged = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(merged) == 1 + assert merged[0].get("sourceError") == "Agent used wrong approach" + + def test_protected_category_rejected_at_apply_time(self): + """P1: correction facts proposed by LLM slip must be rejected at apply time.""" + updater = MemoryUpdater() + # correction category has consolidation_min_facts-1 facts (below threshold), + # but we give the LLM a chance to propose them anyway (simulating a slip). + # We need ≥ consolidation_min_facts correction facts to even appear in + # allowed_source_ids — so we put them BELOW threshold to confirm they're blocked. + correction_facts = [{**_make_fact(f"corr_{i}", f"Correction {i}", "correction", 0.95), "sourceError": "wrong approach"} for i in range(3)] + current_memory = _make_memory(correction_facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["corr_0", "corr_1"], + "consolidated": {"content": "Merged corrections", "category": "correction", "confidence": 0.95}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + consolidation_min_facts=8, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + # All 3 correction facts must survive untouched + assert len(result["facts"]) == 3 + ids = {f["id"] for f in result["facts"]} + assert "corr_0" in ids and "corr_1" in ids and "corr_2" in ids + assert all(f.get("source") != "consolidation" for f in result["facts"]) + + def test_confidence_cap_and_threshold_gate(self): + """P2a: LLM-returned confidence is capped at max source confidence; result below threshold is rejected.""" + updater = MemoryUpdater() + facts = [ + _make_fact("fact_a", "Fact A", "knowledge", 0.75), + _make_fact("fact_b", "Fact B", "knowledge", 0.75), + ] + current_memory = _make_memory(facts) + + # Case 1: LLM returns conf=1.0, sources max at 0.75 → capped to 0.75 + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": {"content": "Merged", "category": "knowledge", "confidence": 1.0}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + fact_confidence_threshold=0.7, + consolidation_min_facts=2, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + merged = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(merged) == 1, "merge should succeed" + assert merged[0]["confidence"] == 0.75, "confidence must be capped at max source confidence" + + # Case 2: sources max at 0.65, below fact_confidence_threshold=0.7 → rejected + facts2 = [ + _make_fact("fact_c", "Fact C", "knowledge", 0.65), + _make_fact("fact_d", "Fact D", "knowledge", 0.60), + ] + current_memory2 = _make_memory(facts2) + update_data2 = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_c", "fact_d"], + "consolidated": {"content": "Below threshold", "category": "knowledge", "confidence": 1.0}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + fact_confidence_threshold=0.7, + consolidation_min_facts=2, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result2 = updater._apply_updates(current_memory2, update_data2) + + # Both source facts must survive untouched — consolidation was rejected + assert len(result2["facts"]) == 2 + assert all(f.get("source") != "consolidation" for f in result2["facts"]) + + def test_apply_gate_consolidation_disabled(self): + """P2b: factsToConsolidate present but consolidation_enabled=False → nothing merged at apply time.""" + updater = MemoryUpdater() + facts = [ + _make_fact("fact_a", "Fact A", "knowledge", 0.9), + _make_fact("fact_b", "Fact B", "knowledge", 0.85), + ] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + "consolidated": {"content": "Should not merge", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=False, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + assert len(result["facts"]) == 2, "both source facts must survive when consolidation is disabled" + assert all(f.get("source") != "consolidation" for f in result["facts"]) + + def test_consolidation_enabled_defaults_to_false(self): + """Finding 1: consolidation is opt-in — default must be False to avoid lossy mutations on first deploy.""" + from deerflow.config.memory_config import MemoryConfig + + assert MemoryConfig().consolidation_enabled is False + + def test_null_confidence_renders_consistently_with_cap(self): + """Finding 2: a fact with confidence=None must show the same value in the prompt as in the confidence cap.""" + null_fact = {**_make_fact("fact_null", "null conf fact", "knowledge"), "confidence": None} + other_fact = _make_fact("fact_b", "normal fact", "knowledge", 0.9) + + # Prompt rendering must use _coerce_source_confidence default (0.5), not 0.0 + section = _build_consolidation_section({"knowledge": [null_fact, other_fact]}) + assert "0.50" in section, "null confidence must render as 0.50 (coerced default), not 0.00" + assert "0.00" not in section + + # Apply-time cap must also use 0.5 for the null-confidence source + updater = MemoryUpdater() + current_memory = _make_memory([null_fact, other_fact]) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_null", "fact_b"], + # LLM returns 1.0; cap = max(0.5, 0.9) = 0.9 + "consolidated": {"content": "Merged", "category": "knowledge", "confidence": 1.0}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + fact_confidence_threshold=0.5, + consolidation_enabled=True, + consolidation_min_facts=2, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + merged = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(merged) == 1, "merge should succeed" + # cap = max(coerce(null)=0.5, coerce(0.9)=0.9) = 0.9; LLM conf 1.0 capped → 0.9 + assert merged[0]["confidence"] == pytest.approx(0.9) + + def test_consolidated_created_at_tracks_newest_source(self): + """Finding 3: createdAt must equal the newest source's createdAt (not now) to preserve staleness eligibility.""" + updater = MemoryUpdater() + older_date = "2025-01-01T00:00:00Z" + newer_date = "2026-03-15T12:00:00Z" + facts = [ + {**_make_fact("fact_old", "Old fact", "knowledge", 0.9), "createdAt": older_date}, + {**_make_fact("fact_new", "New fact", "knowledge", 0.85), "createdAt": newer_date}, + ] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_old", "fact_new"], + "consolidated": {"content": "Old and new merged", "category": "knowledge", "confidence": 0.9}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + consolidation_min_facts=2, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + merged = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(merged) == 1 + # createdAt must be the newest source's date — staleness clock not reset + assert merged[0]["createdAt"] == newer_date, "createdAt must equal newest source's date" + # consolidatedAt must be present as an audit field + assert "consolidatedAt" in merged[0], "consolidatedAt must be set for auditability" + # consolidatedAt should be more recent than the source dates + assert merged[0]["consolidatedAt"] > newer_date + + def test_confidence_fallback_to_max_source_when_llm_omits_field(self): + """Finding 5: when LLM omits confidence field entirely, merged fact uses max_source_conf.""" + updater = MemoryUpdater() + facts = [ + _make_fact("fact_a", "Fact A", "knowledge", 0.85), + _make_fact("fact_b", "Fact B", "knowledge", 0.75), + ] + current_memory = _make_memory(facts) + update_data = { + "user": {}, + "history": {}, + "newFacts": [], + "factsToRemove": [], + "staleFactsToRemove": [], + "factsToConsolidate": [ + { + "sourceIds": ["fact_a", "fact_b"], + # LLM omits the confidence field entirely + "consolidated": {"content": "Merged without confidence", "category": "knowledge"}, + }, + ], + } + with patch( + "deerflow.agents.memory.updater.get_memory_config", + return_value=_memory_config( + max_facts=100, + consolidation_enabled=True, + consolidation_min_facts=2, + consolidation_max_groups_per_cycle=3, + consolidation_max_sources=8, + ), + ): + result = updater._apply_updates(current_memory, update_data) + + merged = [f for f in result["facts"] if f.get("source") == "consolidation"] + assert len(merged) == 1, "merge should succeed" + # fallback: max(coerce(0.85), coerce(0.75)) = 0.85 + assert merged[0]["confidence"] == pytest.approx(0.85) + + +# ── Integration: _prepare_update_prompt ──────────────────────────────────── + + +class TestPrepareUpdatePromptConsolidation: + def test_consolidation_section_included_when_triggered(self): + updater = MemoryUpdater() + facts = [_make_fact(f"fact_{i}", f"Knowledge {i}", "knowledge", 0.8) for i in range(10)] + memory = _make_memory(facts) + + msg = MagicMock() + msg.type = "human" + msg.content = "Hello" + + config = _memory_config( + enabled=True, + consolidation_enabled=True, + consolidation_min_facts=8, + ) + + with ( + patch("deerflow.agents.memory.updater.get_memory_config", return_value=config), + patch("deerflow.agents.memory.updater.get_memory_data", return_value=memory), + ): + result = updater._prepare_update_prompt( + messages=[msg], + agent_name=None, + correction_detected=False, + reinforcement_detected=False, + ) + + assert result is not None + _, prompt = result + assert "Memory Consolidation" in prompt + assert "consolidation_candidates" in prompt + + def test_consolidation_section_omitted_when_not_triggered(self): + updater = MemoryUpdater() + memory = _make_memory([_make_fact("fact_only", category="knowledge")]) + + msg = MagicMock() + msg.type = "human" + msg.content = "Hello" + + config = _memory_config( + enabled=True, + consolidation_enabled=True, + consolidation_min_facts=8, + ) + + with ( + patch("deerflow.agents.memory.updater.get_memory_config", return_value=config), + patch("deerflow.agents.memory.updater.get_memory_data", return_value=memory), + ): + result = updater._prepare_update_prompt( + messages=[msg], + agent_name=None, + correction_detected=False, + reinforcement_detected=False, + ) + + assert result is not None + _, prompt = result + assert "Memory Consolidation" not in prompt + + def test_consolidation_section_omitted_when_disabled(self): + updater = MemoryUpdater() + facts = [_make_fact(f"fact_{i}", category="knowledge") for i in range(20)] + memory = _make_memory(facts) + + msg = MagicMock() + msg.type = "human" + msg.content = "Hello" + + config = _memory_config( + enabled=True, + consolidation_enabled=False, + ) + + with ( + patch("deerflow.agents.memory.updater.get_memory_config", return_value=config), + patch("deerflow.agents.memory.updater.get_memory_data", return_value=memory), + ): + result = updater._prepare_update_prompt( + messages=[msg], + agent_name=None, + correction_detected=False, + reinforcement_detected=False, + ) + + assert result is not None + _, prompt = result + assert "Memory Consolidation" not in prompt diff --git a/config.example.yaml b/config.example.yaml index 20f05a590..3ea98e048 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1438,6 +1438,26 @@ memory: staleness_protected_categories: - correction + # Memory consolidation: when a single category accumulates many fragmented + # facts, the LLM reviews them during the normal memory-update call (same + # invocation — no extra API call) and decides whether groups of related facts + # can be synthesized into a single richer fact. + # consolidation_enabled defaults to false because consolidation is lossy: + # source fact content is permanently replaced by the LLM-synthesized fact + # (only the source IDs are kept in consolidatedFrom). Enable explicitly once + # you are comfortable with that trade-off. + # consolidation_enabled - master switch (default: false) + # consolidation_min_facts - minimum facts in a category to trigger + # consolidation review (default: 8) + # consolidation_max_groups_per_cycle - safety cap on merges per cycle + # (default: 3) + # consolidation_max_sources - max source facts per merge group; + # prevents over-merging (default: 8) + consolidation_enabled: false + consolidation_min_facts: 8 + consolidation_max_groups_per_cycle: 3 + consolidation_max_sources: 8 + # ============================================================================ # Custom Agent Management API # ============================================================================