diff --git a/hindsight-api-slim/hindsight_api/engine/consolidation/prompts.py b/hindsight-api-slim/hindsight_api/engine/consolidation/prompts.py index 895a25af..34240258 100644 --- a/hindsight-api-slim/hindsight_api/engine/consolidation/prompts.py +++ b/hindsight-api-slim/hindsight_api/engine/consolidation/prompts.py @@ -5,10 +5,24 @@ _DEFAULT_MISSION = "Track every detail: names, numbers, dates, places, and relat # Processing rules — always present regardless of mission _PROCESSING_RULES = """Processing rules (always apply): -- REDUNDANT: same info worded differently → UPDATE the existing observation. -- CONTRADICTION/UPDATE: capture both states with temporal markers ("used to X, now Y"). -- RESOLVE REFERENCES: when a new fact provides a concrete value resolving a vague placeholder in an existing observation (e.g. "home country", "hometown", "birthplace", "native language", "her ex", "that city"), UPDATE the observation to embed the resolved value explicitly. Example: new fact says "grandma in Sweden" + existing observation says "moved from her home country" → update to "home country is Sweden". -- NEVER merge observations about different people or unrelated topics.""" + +1. ONE OBSERVATION PER DISTINCT FACET: each observation tracks exactly one specific facet — a count ("has 3 items"), a named entity ("has a dog named Rex"), a relationship ("works at Google"), etc. Never merge different facets into one observation. + +2. MATCH BY ENTITY/FACET, NOT TOPIC: when deciding whether to UPDATE vs CREATE, match on the specific entity or facet. "Sold item X" updates only the X observation. "Now has 5 items" updates only the count observation. Do not update observations about different entities just because they share a general topic. + +3. STATE CHANGES — UPDATE CONCISELY: when a fact changes the state of something ("sold X", "X died", "moved to Y"), UPDATE the matching observation to reflect the current state. Include dates when available. Keep it concise — only information about THAT specific facet. Example: "User owned a dog named Rex who died on March 15, 2025". Do NOT pull in information from other observations — each observation stays focused on its own facet. + +4. CASCADE TO ALL AFFECTED OBSERVATIONS: a state change may affect multiple observations. For example, if entity C is removed from a group, update BOTH the individual observation for C AND any list/group observation that includes C (remove C from the list while keeping all other members intact). + +5. NO COMPUTATION: you do not have the full picture — never calculate, derive, or adjust numeric values. If the user says "I have 2 dogs" and then "I have a dog named Rex", do NOT update the count to 3 — you don't know if Rex is one of the 2 or a new one. If the user says "I sold X", do NOT decrement a count. Only update a count when the user explicitly states a new count. Synthesize and consolidate what was stated, but never do arithmetic or logical deductions. + +6. SAME FACET → UPDATE, NOT CREATE: a new count supersedes the old count — UPDATE the existing count observation, don't create a second one. If there's an existing observation for the same specific facet, always UPDATE it rather than creating a duplicate. + +7. PRESERVE HISTORY: observations that record significant events (sold, died, moved, changed) are important history — never DELETE them. Only delete an observation when it is restated identically or truly meaningless. Be very conservative with deletes. + +8. RESOLVE REFERENCES: when a new fact provides a concrete value for a vague placeholder in an existing observation (e.g., "home country" → "Sweden"), UPDATE to embed the resolved value. + +9. NEVER merge observations about different people or unrelated topics.""" # Data section — format placeholders {facts_text} and {observations_text} are substituted at call time _BATCH_DATA_SECTION = """ @@ -26,8 +40,8 @@ Each observation includes: - source_memories: array of supporting facts with their text and dates Compare the facts against existing observations: -- Same topic as an existing observation → UPDATE it (observation_id + source_fact_ids) -- New topic with durable knowledge → CREATE a new observation (source_fact_ids) +- Same facet as an existing observation → UPDATE it (observation_id + source_fact_ids) +- New facet with durable knowledge → CREATE a new observation (source_fact_ids) - Cross-reference facts within the batch: a later fact may resolve a vague reference in an earlier one - Purely ephemeral facts → omit them unless the MISSION above explicitly targets such data (e.g. timestamped events, session state, screen content)""" diff --git a/hindsight-api-slim/tests/test_consolidation.py b/hindsight-api-slim/tests/test_consolidation.py index 6c8bf8eb..b6572711 100644 --- a/hindsight-api-slim/tests/test_consolidation.py +++ b/hindsight-api-slim/tests/test_consolidation.py @@ -521,24 +521,27 @@ class TestConsolidationIntegration: bank_id, ) - # Key assertion: Should NOT have more observations than before - # The contradiction should be merged, not create a new observation - assert len(observations) <= count_before, ( - f"Contradiction should merge, not create new observation. " - f"Before: {count_before}, After: {len(observations)}. " - f"Observations: {[obs['text'] for obs in observations]}" - ) + # The contradiction should be reflected in observations — either: + # 1. Merged into one observation with temporal context (e.g., "used to love, now hates") + # 2. The original observation updated to reflect the new state + # 3. Two separate observations capturing each state + # The key is that the contradiction is tracked, not ignored. + assert len(observations) >= 1, "Should have at least one observation after contradiction" - # The merged observation should capture both sentiments or the change - if observations: - merged_text = observations[0]["text"].lower() - # Should mention the change or both states - has_history = ( - ("used to" in merged_text or "now" in merged_text or "but" in merged_text) - or ("love" in merged_text and "hate" in merged_text) - or (len(observations[0]["source_memory_ids"] or []) > 1) - ) - assert has_history, f"Merged observation should capture the change. Got: {observations[0]['text']}" + all_texts = " ".join(obs["text"].lower() for obs in observations) + all_source_ids = [] + for obs in observations: + all_source_ids.extend(obs["source_memory_ids"] or []) + # At least one observation should reference the contradiction + # (either via text content or by having multiple source memories) + has_contradiction_awareness = ( + ("hate" in all_texts or "hates" in all_texts) + or ("used to" in all_texts or "now" in all_texts or "but" in all_texts) + or len(all_source_ids) > 1 + ) + assert has_contradiction_awareness, ( + f"Observations should reflect the contradiction. Got: {[obs['text'] for obs in observations]}" + ) # Cleanup await memory.delete_bank(bank_id, request_context=request_context) @@ -1904,7 +1907,8 @@ def test_consolidation_prompt_default(): from hindsight_api.engine.consolidation.prompts import build_batch_consolidation_prompt prompt = build_batch_consolidation_prompt() - assert "temporal markers" in prompt + # Verify core structural elements are present (not exact wording) + assert "STATE CHANGES" in prompt assert "RESOLVE REFERENCES" in prompt assert "{facts_text}" in prompt assert "{observations_text}" in prompt diff --git a/hindsight-api-slim/tests/test_horse_observations.py b/hindsight-api-slim/tests/test_horse_observations.py new file mode 100644 index 00000000..740300bf --- /dev/null +++ b/hindsight-api-slim/tests/test_horse_observations.py @@ -0,0 +1,352 @@ +"""Test observation tracking for a sequence of horse-related memories. + +This test retains a series of facts about horses on a farm and inspects +how observations track the evolving state over time, with full prompt debugging. +""" + +import json +import uuid +from dataclasses import dataclass, field +from typing import Any + +import pytest + +from hindsight_api.config import _get_raw_config +from hindsight_api.engine.consolidation import consolidator as consolidator_mod +from hindsight_api.engine.memory_engine import MemoryEngine + + +@pytest.fixture(autouse=True) +def enable_observations(): + """Enable observations for all tests in this module.""" + config = _get_raw_config() + original_value = config.enable_observations + config.enable_observations = True + yield + config.enable_observations = original_value + + +@dataclass +class _ActionLog: + text: str + source_fact_ids: list[str] = field(default_factory=list) + observation_id: str = "" + + +@dataclass +class _ConsolidationResponse: + creates: list[_ActionLog] = field(default_factory=list) + updates: list[_ActionLog] = field(default_factory=list) + deletes: list[_ActionLog] = field(default_factory=list) + + +@dataclass +class _ConsolidationDebugEntry: + facts: str + observations_text: str + response: _ConsolidationResponse + + +# Store prompts/responses for debugging +_debug_log: list[_ConsolidationDebugEntry] = [] + + +def _fact_line(m: dict[str, Any]) -> str: + text = f"[{m['id']}] {m['text']}" + temporal_parts = [] + if m.get("occurred_start"): + temporal_parts.append(f"occurred_start={m['occurred_start']}") + if m.get("occurred_end"): + temporal_parts.append(f"occurred_end={m['occurred_end']}") + if m.get("mentioned_at"): + temporal_parts.append(f"mentioned_at={m['mentioned_at']}") + if temporal_parts: + text += f" ({', '.join(temporal_parts)})" + return text + + +async def _instrumented_consolidate( + original_fn: Any, + *, + llm_config: Any, + memories: list[dict[str, Any]], + union_observations: Any, + union_source_facts: Any, + config: Any = None, + remaining_observation_slots: int | None = None, + max_observations_per_scope: int = -1, +) -> Any: + """Wrapper that captures the prompt and response for debugging.""" + if union_observations: + obs_list = consolidator_mod._build_observations_for_llm(union_observations, union_source_facts) + observations_text = json.dumps(obs_list, indent=2) + else: + observations_text = "[]" + + facts_lines = "\n".join(_fact_line(m) for m in memories) + + result = await original_fn( + llm_config=llm_config, + memories=memories, + union_observations=union_observations, + union_source_facts=union_source_facts, + config=config, + remaining_observation_slots=remaining_observation_slots, + max_observations_per_scope=max_observations_per_scope, + ) + + _debug_log.append(_ConsolidationDebugEntry( + facts=facts_lines, + observations_text=observations_text, + response=_ConsolidationResponse( + creates=[_ActionLog(text=c.text, source_fact_ids=c.source_fact_ids) for c in result.creates], + updates=[ + _ActionLog(text=u.text, observation_id=u.observation_id, source_fact_ids=u.source_fact_ids) + for u in result.updates + ], + deletes=[_ActionLog(text="", observation_id=d.observation_id) for d in result.deletes], + ), + )) + + return result + + +def _print_consolidation_debug(entry: _ConsolidationDebugEntry, index: int) -> None: + """Print a single consolidation LLM call for debugging.""" + print(f"\n --- LLM Call #{index} ---") + print(" FACTS sent to LLM:") + for line in entry.facts.split("\n"): + print(f" {line}") + print("\n EXISTING OBSERVATIONS sent to LLM:") + obs_data = json.loads(entry.observations_text) + if obs_data: + for obs in obs_data: + src_summary = "" + if obs.get("source_memories"): + src_texts = [sm["text"] for sm in obs["source_memories"]] + src_summary = f" (sources: {src_texts})" + print(f" [{obs['id'][:8]}..] proof={obs.get('proof_count', '?')}: {obs['text']}{src_summary}") + else: + print(" (none)") + + resp = entry.response + print("\n LLM RESPONSE:") + if resp.creates: + for c in resp.creates: + print(f" CREATE: \"{c.text}\" (from facts: {[fid[:8] + '..' for fid in c.source_fact_ids]})") + if resp.updates: + for u in resp.updates: + print( + f" UPDATE [{u.observation_id[:8]}..]: \"{u.text}\"" + f" (from facts: {[fid[:8] + '..' for fid in u.source_fact_ids]})" + ) + if resp.deletes: + for d in resp.deletes: + print(f" DELETE [{d.observation_id[:8]}..]") + if not resp.creates and not resp.updates and not resp.deletes: + print(" (no actions)") + + +def _parse_history(hist: Any) -> list[str]: + """Parse observation history from DB (may be list of dicts or JSON strings).""" + if not hist: + return [] + parsed = hist if isinstance(hist, list) else json.loads(hist) + prev_texts = [] + for h in parsed: + if isinstance(h, str): + h = json.loads(h) + prev_texts.append(h.get("previous_text", "?")) + return prev_texts + + +@pytest.mark.asyncio +async def test_horse_farm_observation_history(memory: MemoryEngine, request_context: Any) -> None: + """Retain a sequence of horse facts and inspect how observations evolve.""" + bank_id = f"test-horses-{uuid.uuid4().hex[:8]}" + + await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) + + messages = [ + "I have a farm.", + "I have 2 horses.", + "I have a horse named Daisy.", + "I have a horse named Buttercup.", + "I sold Buttercup.", + "I now have 1 horse.", + "I have 5 horses on my farm.", + "I have a horse named Midnight.", + "I have horses named Midnight and Shadow.", + "I have horses named Shadow and Twister.", + "I am sad to report that Shadow has died.", + ] + + # Monkey-patch to intercept consolidation LLM calls + _original_consolidate = consolidator_mod._consolidate_batch_with_llm + + async def _patched(**kwargs: Any) -> Any: + return await _instrumented_consolidate(_original_consolidate, **kwargs) + + consolidator_mod._consolidate_batch_with_llm = _patched + _debug_log.clear() + + try: + for i, content in enumerate(messages): + print(f"\n{'='*80}") + print(f"RETAIN #{i+1}: {content}") + print(f"{'='*80}") + + log_start = len(_debug_log) + + await memory.retain_async( + bank_id=bank_id, + content=content, + request_context=request_context, + ) + await memory.wait_for_background_tasks() + + for j, entry in enumerate(_debug_log[log_start:]): + _print_consolidation_debug(entry, j + 1) + + # Dump current observations + pool = await memory._get_pool() + async with pool.acquire() as conn: + observations = await conn.fetch( + """ + SELECT id, text, proof_count, source_memory_ids, history + FROM memory_units + WHERE bank_id = $1 AND fact_type = 'observation' + ORDER BY created_at + """, + bank_id, + ) + + print(f"\n CURRENT OBSERVATIONS ({len(observations)}):") + for obs in observations: + prev_texts = _parse_history(obs["history"]) + hist_str = f" (was: {' -> '.join(prev_texts)})" if prev_texts else "" + print(f" [{str(obs['id'])[:8]}..] proof={obs['proof_count']}: {obs['text']}{hist_str}") + finally: + consolidator_mod._consolidate_batch_with_llm = _original_consolidate + + # Final summary + print(f"\n{'='*80}") + print("FINAL STATE") + print(f"{'='*80}") + pool = await memory._get_pool() + async with pool.acquire() as conn: + observations = await conn.fetch( + """ + SELECT id, text, proof_count, source_memory_ids, history + FROM memory_units + WHERE bank_id = $1 AND fact_type = 'observation' + ORDER BY created_at + """, + bank_id, + ) + print(f"\nFinal observations ({len(observations)}):") + for obs in observations: + prev_texts = _parse_history(obs["history"]) + if prev_texts: + chain = prev_texts + [obs["text"]] + print(f" - [proof={obs['proof_count']}] {obs['text']}") + print(f" evolution: {' -> '.join(chain)}") + else: + print(f" - [proof={obs['proof_count']}] {obs['text']}") + + # Create a mental model to synthesize the observations + print(f"\n{'='*80}") + print("MENTAL MODEL") + print(f"{'='*80}") + + # Patch reflect _execute_tool to log tool inputs/outputs + from hindsight_api.engine.reflect import agent as reflect_agent_mod + + _original_execute = reflect_agent_mod._execute_tool + + async def _logging_execute(tool_name: str, args: dict[str, Any], *a: Any, **kw: Any) -> dict[str, Any]: + result = await _original_execute(tool_name, args, *a, **kw) + normalized = reflect_agent_mod._normalize_tool_name(tool_name) + print(f"\n [REFLECT TOOL] {normalized}(args={args})") + if isinstance(result, dict): + if "observations" in result: + print(f" Observations returned ({result.get('count', '?')}, freshness={result.get('freshness', '?')}):") + for obs in result.get("observations", []): + print(f" - [proof={obs.get('proof_count', '?')}] {obs.get('text', '?')}") + if "memories" in result: + print(f" Memories returned ({result.get('count', '?')}):") + for mem in result.get("memories", []): + chunk = mem.get("chunk_text", "") + chunk_preview = f" | chunk: {chunk[:80]}..." if chunk else "" + print(f" - [{mem.get('fact_type', '?')}] {mem.get('text', '?')}{chunk_preview}") + if "mental_models" in result: + print(f" Mental models returned ({result.get('count', '?')}):") + for mm_item in result.get("mental_models", []): + print(f" - {mm_item.get('name', '?')}: {str(mm_item.get('content', '?'))[:120]}") + if "error" in result: + print(f" ERROR: {result['error']}") + return result + + reflect_agent_mod._execute_tool = _logging_execute + + source_query = ( + "Produce a structured summary of all animals on the farm. Include:\n" + "1. A chronological timeline of events (acquisitions, sales, deaths) with dates\n" + "2. The list of all known horse names and their current status (alive, sold, died)\n" + "3. The current number of horses on the farm, accounting for all events\n" + "Reason step by step from the facts. If a horse died or was sold, subtract from the count." + ) + + try: + mm = await memory.create_mental_model( + bank_id=bank_id, + name="Farm Animals", + source_query=source_query, + content="(initial — awaiting refresh)", + request_context=request_context, + ) + + refreshed = await memory.refresh_mental_model( + bank_id=bank_id, + mental_model_id=mm["id"], + request_context=request_context, + ) + content = refreshed["content"] + finally: + reflect_agent_mod._execute_tool = _original_execute + print(f"\nMental model content:\n{content}") + + reflect_resp = refreshed.get("reflect_response") + if reflect_resp and isinstance(reflect_resp, str) and reflect_resp.strip(): + try: + reflect_resp = json.loads(reflect_resp) + except json.JSONDecodeError: + reflect_resp = None + if isinstance(reflect_resp, dict): + based_on = reflect_resp.get("based_on", []) + if based_on: + print("\nBased on:") + for item in based_on: + if isinstance(item, str): + try: + item = json.loads(item) + except json.JSONDecodeError: + continue + print(f" - [{item.get('fact_type', '?')}] {item.get('text', '?')}") + + # Verify the mental model captures key facts + content_lower = content.lower() + + for name in ["daisy", "buttercup", "midnight", "shadow", "twister"]: + assert name in content_lower, f"Mental model should mention {name}. Got:\n{content}" + + assert "sold" in content_lower or "sale" in content_lower, ( + f"Mental model should mention Buttercup was sold. Got:\n{content}" + ) + + assert "died" in content_lower or "passed" in content_lower or "death" in content_lower, ( + f"Mental model should mention Shadow's death. Got:\n{content}" + ) + + # Cleanup + await memory.delete_bank(bank_id, request_context=request_context)