feat(history): move mental-model & observation history into dedicated tables (#2007)

Both histories accumulated in a single JSONB/CLOB `history` column, appended to on every update. Observations had NO cap at all, so a frequently-reinforced observation grew until it crossed Postgres's 256MB jsonb limit (SQLSTATE 54000) and the row got stuck. Mental models capped by entry COUNT (not size) and rewrote the whole array + TOAST per refresh, defeating HOT updates. Now one row per change in mental_model_history / observation_history, indexed on (item, changed_at DESC, id DESC). Each row stores its snapshot as a single JSONB `content` blob (per-row, so it stays small) plus changed_at; the cap is enforced at write time as a bounded DELETE of the oldest over-cap rows, for both histories (new per-observation cap: HINDSIGHT_API_OBSERVATION_HISTORY_MAX_ENTRIES, default 50). - migration a7b8c9d0e1f2: create tables, backfill from the JSONB/CLOB arrays (PG jsonb_array_elements / Oracle JSON_TABLE), drop the legacy columns - write paths: insert-then-trim in consolidator (observations) and memory_engine (mental models); also stop writing the dropped column in the create-observation INSERT - read paths: get_observation_history / get_mental_model_history read the new tables; observation list/get no longer select the column - export/import: mental_model_history carried (parent keeps a stable id, the surrogate id is dropped so the target reassigns it); observation_history is derived (observations regenerate with fresh ids on import) and not carried - tests: deterministic observation-history coverage + MM-history export/import round-trip
2026-09-14 19:31:49 +08:00 · 2026-06-05 15:07:30 +02:00
parent 75a7c19d6a
commit 7e1145c08a
16 changed files with 707 additions and 143 deletions
@@ -52,7 +52,9 @@ BACKUP_TABLES = [
    "unit_entities",
    "entity_cooccurrences",
    "memory_links",
+    "observation_history",
    "mental_models",
+    "mental_model_history",
    "directives",
    "async_operations",
    "webhooks",
@@ -0,0 +1,253 @@
+"""Move mental-model and observation history into dedicated tables.
+
+Both histories were accumulated in a single JSONB/CLOB ``history`` column
+(``mental_models.history`` and ``memory_units.history``), appended to on every
+update. That design has two problems:
+
+1. **Unbounded growth on observations.** The observation write path appended a
+   snapshot on every update with no cap at all, so a frequently-reinforced
+   observation grew its ``history`` array until it crossed Postgres's hard 256MB
+   jsonb limit (SQLSTATE 54000), after which every further UPDATE failed and the
+   row was stuck.
+2. **Wrong-axis cap on mental models.** The mental-model cap bounded the *number*
+   of entries (50), not their *size* — a single large reflect snapshot could
+   still blow the budget — and rewrote the whole array (plus TOAST) on every
+   refresh, defeating HOT updates.
+
+This migration creates one row per history entry in two dedicated tables, with
+an index that makes "most recent N for this item" cheap, then drops the old
+columns. The cap is now enforced at write time as a bounded DELETE of the
+oldest over-cap rows (see config ``*_HISTORY_MAX_ENTRIES``).
+
+Revision ID: a7b8c9d0e1f2
+Revises: d3e4f5a6b7c8
+Create Date: 2026-06-05
+"""
+
+from collections.abc import Sequence
+
+from alembic import context, op
+
+from hindsight_api.alembic._dialect import run_for_dialect
+
+revision: str = "a7b8c9d0e1f2"
+down_revision: str | Sequence[str] | None = "d3e4f5a6b7c8"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _get_schema_prefix() -> str:
+    """Schema-qualifier for raw SQL on PG (multi-tenant search_path)."""
+    schema = context.config.get_main_option("target_schema")
+    return f'"{schema}".' if schema else ""
+
+
+# ---------------------------------------------------------------------------
+# PostgreSQL
+# ---------------------------------------------------------------------------
+def _pg_upgrade() -> None:
+    schema = _get_schema_prefix()
+
+    # Both tables share the same shape: surrogate id, FK to the parent, bank_id,
+    # the snapshot payload as a single JSONB ``content`` blob, and changed_at.
+    # The payload is per-row (one change per row) so it stays small — this is NOT
+    # the old single-column-grows-forever design; growth is bounded by row count
+    # plus the write-time cap. Folding the previous_* fields into one JSONB keeps
+    # the schema dialect-simple (no array columns) and flexible.
+
+    # --- mental_model_history -------------------------------------------------
+    # content: {"previous_content": ..., "previous_reflect_response": {...}}
+    op.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS {schema}mental_model_history (
+            id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+            mental_model_id VARCHAR(64) NOT NULL,
+            bank_id VARCHAR(64) NOT NULL,
+            content JSONB NOT NULL,
+            changed_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(),
+            FOREIGN KEY (mental_model_id, bank_id)
+                REFERENCES {schema}mental_models(id, bank_id) ON DELETE CASCADE
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS idx_mm_history_model "
+        f"ON {schema}mental_model_history (bank_id, mental_model_id, changed_at DESC, id DESC)"
+    )
+
+    # --- observation_history --------------------------------------------------
+    # content: {"previous_text", "previous_tags", "previous_occurred_start",
+    #           "previous_occurred_end", "previous_mentioned_at", "new_source_memory_ids"}
+    op.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS {schema}observation_history (
+            id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+            observation_id UUID NOT NULL,
+            bank_id VARCHAR(64) NOT NULL,
+            content JSONB NOT NULL,
+            changed_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(),
+            FOREIGN KEY (observation_id)
+                REFERENCES {schema}memory_units(id) ON DELETE CASCADE
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX IF NOT EXISTS idx_observation_history_obs "
+        f"ON {schema}observation_history (observation_id, changed_at DESC, id DESC)"
+    )
+
+    # --- backfill mental models ----------------------------------------------
+    # Explode each row's history array into rows, preserving chronological order
+    # via WITH ORDINALITY so the IDENTITY id tie-breaks oldest->newest correctly.
+    # changed_at is promoted to its own column; the rest of the element becomes
+    # ``content`` (the ``- 'changed_at'`` strips the now-redundant key).
+    op.execute(
+        f"""
+        INSERT INTO {schema}mental_model_history (mental_model_id, bank_id, content, changed_at)
+        SELECT mm.id, mm.bank_id,
+               e - 'changed_at',
+               COALESCE(NULLIF(e->>'changed_at', '')::timestamptz, now())
+        FROM {schema}mental_models mm
+             CROSS JOIN LATERAL jsonb_array_elements(mm.history) WITH ORDINALITY a(e, ord)
+        WHERE mm.history IS NOT NULL
+          AND jsonb_typeof(mm.history) = 'array'
+          AND jsonb_array_length(mm.history) > 0
+        ORDER BY mm.id, mm.bank_id, ord
+        """
+    )
+
+    # --- backfill observations -----------------------------------------------
+    op.execute(
+        f"""
+        INSERT INTO {schema}observation_history (observation_id, bank_id, content, changed_at)
+        SELECT mu.id, mu.bank_id,
+               e - 'changed_at',
+               COALESCE(NULLIF(e->>'changed_at', '')::timestamptz, now())
+        FROM {schema}memory_units mu
+             CROSS JOIN LATERAL jsonb_array_elements(mu.history) WITH ORDINALITY a(e, ord)
+        WHERE mu.fact_type = 'observation'
+          AND mu.history IS NOT NULL
+          AND jsonb_typeof(mu.history) = 'array'
+          AND jsonb_array_length(mu.history) > 0
+        ORDER BY mu.id, ord
+        """
+    )
+
+    # --- drop the legacy columns ---------------------------------------------
+    op.execute(f"ALTER TABLE {schema}mental_models DROP COLUMN IF EXISTS history")
+    op.execute(f"ALTER TABLE {schema}memory_units DROP COLUMN IF EXISTS history")
+
+
+def _pg_downgrade() -> None:
+    schema = _get_schema_prefix()
+
+    # Re-add the columns (empty — historical content is not reconstructed back
+    # into the array form; the dedicated tables are dropped below).
+    op.execute(f"ALTER TABLE {schema}mental_models ADD COLUMN IF NOT EXISTS history JSONB DEFAULT '[]'::jsonb")
+    op.execute(f"ALTER TABLE {schema}memory_units ADD COLUMN IF NOT EXISTS history JSONB DEFAULT '[]'::jsonb")
+
+    op.execute(f"DROP INDEX IF EXISTS {schema}idx_observation_history_obs")
+    op.execute(f"DROP TABLE IF EXISTS {schema}observation_history")
+    op.execute(f"DROP INDEX IF EXISTS {schema}idx_mm_history_model")
+    op.execute(f"DROP TABLE IF EXISTS {schema}mental_model_history")
+
+
+# ---------------------------------------------------------------------------
+# Oracle 23ai
+# ---------------------------------------------------------------------------
+def _oracle_upgrade() -> None:
+    # Same single-JSONB shape as PG: ``content`` holds the snapshot payload as a
+    # CLOB IS JSON. The legacy per-element JSON object (minus changed_at, promoted
+    # to its own column) is carried through verbatim on backfill — the array
+    # columns the previous design needed are gone.
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS mental_model_history (
+            id NUMBER GENERATED BY DEFAULT AS IDENTITY,
+            mental_model_id VARCHAR2(256) NOT NULL,
+            bank_id VARCHAR2(256) NOT NULL,
+            content CLOB NOT NULL
+                CONSTRAINT mmh_content_json CHECK (content IS JSON),
+            changed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
+            CONSTRAINT pk_mental_model_history PRIMARY KEY (id),
+            CONSTRAINT fk_mmh_model FOREIGN KEY (mental_model_id, bank_id)
+                REFERENCES mental_models(id, bank_id) ON DELETE CASCADE
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX idx_mm_history_model ON mental_model_history (bank_id, mental_model_id, changed_at DESC, id DESC)"
+    )
+
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS observation_history (
+            id NUMBER GENERATED BY DEFAULT AS IDENTITY,
+            observation_id RAW(16) NOT NULL,
+            bank_id VARCHAR2(256) NOT NULL,
+            content CLOB NOT NULL
+                CONSTRAINT oh_content_json CHECK (content IS JSON),
+            changed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
+            CONSTRAINT pk_observation_history PRIMARY KEY (id),
+            CONSTRAINT fk_oh_obs FOREIGN KEY (observation_id)
+                REFERENCES memory_units(id) ON DELETE CASCADE
+        )
+        """
+    )
+    op.execute(
+        "CREATE INDEX idx_observation_history_obs ON observation_history (observation_id, changed_at DESC, id DESC)"
+    )
+
+    bind = op.get_bind()
+
+    # Backfill via JSON_TABLE. ``content`` is the whole element (FORMAT JSON PATH
+    # '$'); changed_at is also promoted to its own column. Backfilled content may
+    # therefore still carry a redundant changed_at key, which the read path
+    # ignores in favour of the column — harmless, and avoids JSON surgery here.
+    bind.exec_driver_sql(
+        """
+        INSERT INTO mental_model_history (mental_model_id, bank_id, content, changed_at)
+        SELECT mm.id, mm.bank_id, jt.content, NVL(jt.changed_at, SYSTIMESTAMP)
+        FROM mental_models mm,
+             JSON_TABLE(mm.history, '$[*]' COLUMNS (
+                 seq FOR ORDINALITY,
+                 content CLOB FORMAT JSON PATH '$',
+                 changed_at TIMESTAMP WITH TIME ZONE PATH '$.changed_at'
+             )) jt
+        WHERE mm.history IS NOT NULL
+        ORDER BY mm.id, mm.bank_id, jt.seq
+        """
+    )
+
+    bind.exec_driver_sql(
+        """
+        INSERT INTO observation_history (observation_id, bank_id, content, changed_at)
+        SELECT mu.id, mu.bank_id, jt.content, NVL(jt.changed_at, SYSTIMESTAMP)
+        FROM memory_units mu,
+             JSON_TABLE(mu.history, '$[*]' COLUMNS (
+                 seq FOR ORDINALITY,
+                 content CLOB FORMAT JSON PATH '$',
+                 changed_at TIMESTAMP WITH TIME ZONE PATH '$.changed_at'
+             )) jt
+        WHERE mu.fact_type = 'observation' AND mu.history IS NOT NULL
+        ORDER BY mu.id, jt.seq
+        """
+    )
+
+    op.execute("ALTER TABLE mental_models DROP COLUMN history")
+    op.execute("ALTER TABLE memory_units DROP COLUMN history")
+
+
+def _oracle_downgrade() -> None:
+    op.execute("ALTER TABLE mental_models ADD history CLOB DEFAULT '[]' NOT NULL")
+    op.execute("ALTER TABLE memory_units ADD history CLOB DEFAULT '[]'")
+    op.execute("DROP TABLE observation_history CASCADE CONSTRAINTS")
+    op.execute("DROP TABLE mental_model_history CASCADE CONSTRAINTS")
+
+
+def upgrade() -> None:
+    run_for_dialect(pg=_pg_upgrade, oracle=_oracle_upgrade)
+
+
+def downgrade() -> None:
+    run_for_dialect(pg=_pg_downgrade, oracle=_oracle_downgrade)
@@ -446,6 +446,7 @@ ENV_CONSOLIDATION_MAX_ATTEMPTS = "HINDSIGHT_API_CONSOLIDATION_MAX_ATTEMPTS"
 ENV_OBSERVATIONS_MISSION = "HINDSIGHT_API_OBSERVATIONS_MISSION"
 ENV_MAX_OBSERVATIONS_PER_SCOPE = "HINDSIGHT_API_MAX_OBSERVATIONS_PER_SCOPE"
 ENV_ENABLE_OBSERVATION_HISTORY = "HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY"
+ENV_OBSERVATION_HISTORY_MAX_ENTRIES = "HINDSIGHT_API_OBSERVATION_HISTORY_MAX_ENTRIES"
 ENV_ENABLE_MENTAL_MODEL_HISTORY = "HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY"
 ENV_MENTAL_MODEL_HISTORY_MAX_ENTRIES = "HINDSIGHT_API_MENTAL_MODEL_HISTORY_MAX_ENTRIES"

@@ -829,12 +830,16 @@ DEFAULT_ENABLE_OBSERVATIONS = True  # Observations enabled by default
 DEFAULT_ENABLE_AUTO_CONSOLIDATION = True  # Auto-consolidation after retain enabled by default
 DEFAULT_ENABLE_OBSERVATION_HISTORY = True  # Observation history tracking enabled by default
 DEFAULT_ENABLE_MENTAL_MODEL_HISTORY = True  # Mental model history tracking enabled by default
-# Each history entry snapshots previous_content + previous_reflect_response. Without
-# a cap, sustained mental-model refresh load grows the jsonb array unboundedly until
-# it crosses Postgres's hard 256MB jsonb limit and subsequent UPDATEs fail with
-# SQLSTATE 54000. 50 keeps the array well under 100MB even with large reflect
-# responses, while preserving enough recent history for meaningful audit / rollback.
+# History (mental-model refresh snapshots and observation update snapshots) lives in
+# the dedicated mental_model_history / observation_history tables, one row per change.
+# On every write we insert the new entry and delete the oldest rows beyond the cap,
+# so the per-item history can never grow unboundedly (the old single-JSONB-column
+# design hit Postgres's hard 256MB jsonb limit -> SQLSTATE 54000 and stuck rows).
+# 50 preserves enough recent history for meaningful audit / rollback per item.
+# A cap <= 0 removes the trim (unbounded growth) — to turn history OFF use the
+# enable_* flag, not a zero cap.
 DEFAULT_MENTAL_MODEL_HISTORY_MAX_ENTRIES = 50
+DEFAULT_OBSERVATION_HISTORY_MAX_ENTRIES = 50
 DEFAULT_CONSOLIDATION_MAX_ATTEMPTS = 3  # Outer retry attempts for consolidation LLM batch calls
 DEFAULT_CONSOLIDATION_BATCH_SIZE = 50  # Memories to load per batch (internal memory optimization)
 DEFAULT_CONSOLIDATION_MAX_MEMORIES_PER_ROUND = (
@@ -1418,6 +1423,7 @@ class HindsightConfig:
    enable_observations: bool
    enable_auto_consolidation: bool
    enable_observation_history: bool
+    observation_history_max_entries: int
    enable_mental_model_history: bool
    mental_model_history_max_entries: int
    consolidation_batch_size: int
@@ -2278,6 +2284,12 @@ class HindsightConfig:
                ENV_ENABLE_OBSERVATION_HISTORY, str(DEFAULT_ENABLE_OBSERVATION_HISTORY)
            ).lower()
            == "true",
+            observation_history_max_entries=int(
+                os.getenv(
+                    ENV_OBSERVATION_HISTORY_MAX_ENTRIES,
+                    str(DEFAULT_OBSERVATION_HISTORY_MAX_ENTRIES),
+                )
+            ),
            enable_mental_model_history=os.getenv(
                ENV_ENABLE_MENTAL_MODEL_HISTORY, str(DEFAULT_ENABLE_MENTAL_MODEL_HISTORY)
            ).lower()
@@ -22,7 +22,7 @@ import time
 import uuid
 from collections import defaultdict
 from contextlib import AsyncExitStack
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from itertools import combinations
 from typing import TYPE_CHECKING, Any, Literal
@@ -1520,6 +1520,64 @@ def _max_date(dates: "Any") -> "datetime | None":
    return max((d for d in dates if d is not None), default=None)


+@dataclass(frozen=True)
+class _ObservationHistorySnapshot:
+    """Pre-update state of an observation, persisted as the ``content`` JSON blob
+    of one observation_history row.
+
+    Temporal fields are the ISO strings carried on MemoryFact; new_source_memory_ids
+    are the ids added by the update.
+    """
+
+    previous_text: str | None
+    previous_tags: list[str]
+    previous_occurred_start: str | None
+    previous_occurred_end: str | None
+    previous_mentioned_at: str | None
+    new_source_memory_ids: list[str]
+
+
+async def _append_observation_history(
+    conn: "Connection",
+    bank_id: str,
+    observation_id: str,
+    snapshot: _ObservationHistorySnapshot,
+    max_entries: int,
+) -> None:
+    """Insert one pre-update snapshot into ``observation_history``, then delete the
+    oldest rows beyond ``max_entries`` for this observation.
+
+    The snapshot is stored as a single JSONB ``content`` blob (per-row, so it stays
+    small). Bounding by row count keeps a frequently-reinforced observation's
+    history from growing without bound.
+    """
+    obs_uuid = uuid.UUID(observation_id)
+    await conn.execute(
+        f"""
+        INSERT INTO {fq_table("observation_history")} (observation_id, bank_id, content, changed_at)
+        VALUES ($1, $2, $3::jsonb, now())
+        """,
+        obs_uuid,
+        bank_id,
+        json.dumps(asdict(snapshot)),
+    )
+    if max_entries and max_entries > 0:
+        await conn.execute(
+            f"""
+            DELETE FROM {fq_table("observation_history")}
+            WHERE observation_id = $1
+              AND id NOT IN (
+                  SELECT id FROM {fq_table("observation_history")}
+                  WHERE observation_id = $1
+                  ORDER BY changed_at DESC, id DESC
+                  LIMIT $2
+              )
+            """,
+            obs_uuid,
+            max_entries,
+        )
+
+
 async def _execute_update_action(
    conn: "Connection",
    memory_engine: "MemoryEngine",
@@ -1559,15 +1617,14 @@ async def _execute_update_action(

    from ...config import get_config

-    history_entry = {
-        "previous_text": model.text,
-        "previous_tags": list(model.tags or []),
-        "previous_occurred_start": model.occurred_start,
-        "previous_occurred_end": model.occurred_end,
-        "previous_mentioned_at": model.mentioned_at,
-        "changed_at": datetime.now(timezone.utc).isoformat(),
-        "new_source_memory_ids": [str(mid) for mid in source_memory_ids],
-    }
+    history_entry = _ObservationHistorySnapshot(
+        previous_text=model.text,
+        previous_tags=list(model.tags or []),
+        previous_occurred_start=model.occurred_start,
+        previous_occurred_end=model.occurred_end,
+        previous_mentioned_at=model.mentioned_at,
+        new_source_memory_ids=[str(mid) for mid in source_memory_ids],
+    )

    source_ids = list(model.source_fact_ids or []) + source_memory_ids

@@ -1583,9 +1640,6 @@ async def _execute_update_action(
        perf.record_timing("embedding", time.time() - t0)

    config = get_config()
-    history_clause = (
-        "history = COALESCE(history, '[]'::jsonb) || $3::jsonb," if config.enable_observation_history else ""
-    )

    t0 = time.time()
    await conn.execute(
@@ -1593,19 +1647,17 @@ async def _execute_update_action(
        UPDATE {fq_table("memory_units")}
        SET text = $1,
            embedding = $2::vector,
-            {history_clause}
-            source_memory_ids = $4,
-            proof_count = $5,
-            tags = $10,
+            source_memory_ids = $3,
+            proof_count = $4,
+            tags = $9,
            updated_at = now(),
-            occurred_start = LEAST(occurred_start, COALESCE($7, occurred_start)),
-            occurred_end = GREATEST(occurred_end, COALESCE($8, occurred_end)),
-            mentioned_at = GREATEST(mentioned_at, COALESCE($9, mentioned_at))
-        WHERE id = $6
+            occurred_start = LEAST(occurred_start, COALESCE($6, occurred_start)),
+            occurred_end = GREATEST(occurred_end, COALESCE($7, occurred_end)),
+            mentioned_at = GREATEST(mentioned_at, COALESCE($8, mentioned_at))
+        WHERE id = $5
        """,
        new_text,
        embedding_str,
-        json.dumps([history_entry]),
        source_ids,
        len(source_ids),
        uuid.UUID(observation_id),
@@ -1615,6 +1667,15 @@ async def _execute_update_action(
        merged_tags,
    )

+    # Record the pre-update snapshot in the dedicated observation_history table
+    # (one row per change), then trim to the configured cap. History lived in a
+    # single unbounded JSONB column before; an often-reinforced observation grew
+    # it until it crossed Postgres's 256MB jsonb limit and got stuck.
+    if config.enable_observation_history:
+        await _append_observation_history(
+            conn, bank_id, observation_id, history_entry, config.observation_history_max_entries
+        )
+
    # Sync observation_sources junction table (Oracle only — PG uses native array ops).
    if memory_engine._backend.ops.uses_observation_sources_table:
        obs_uuid = uuid.UUID(observation_id)
@@ -2055,10 +2116,10 @@ async def _create_observation_directly(
        # VectorChord: manually tokenize and insert search_vector
        query = f"""
            INSERT INTO {fq_table("memory_units")} (
-                id, bank_id, text, fact_type, embedding, proof_count, source_memory_ids, history,
+                id, bank_id, text, fact_type, embedding, proof_count, source_memory_ids,
                tags, event_date, occurred_start, occurred_end, mentioned_at, search_vector
            )
-            VALUES ($1, $2, $3, 'observation', $4::vector, 1, $5, '[]'::jsonb, $6, $7, $8, $9, $10,
+            VALUES ($1, $2, $3, 'observation', $4::vector, 1, $5, $6, $7, $8, $9, $10,
                    tokenize($3, 'llmlingua2')::bm25_catalog.bm25vector)
            RETURNING id
        """
@@ -2074,10 +2135,10 @@ async def _create_observation_directly(
        # re-ingested. Tracking a separate fix for that gap.
        query = f"""
            INSERT INTO {fq_table("memory_units")} (
-                id, bank_id, text, fact_type, embedding, proof_count, source_memory_ids, history,
+                id, bank_id, text, fact_type, embedding, proof_count, source_memory_ids,
                tags, event_date, occurred_start, occurred_end, mentioned_at
            )
-            VALUES ($1, $2, $3, 'observation', $4::vector, 1, $5, '[]'::jsonb, $6, $7, $8, $9, $10)
+            VALUES ($1, $2, $3, 'observation', $4::vector, 1, $5, $6, $7, $8, $9, $10)
            RETURNING id
        """

@@ -156,6 +156,9 @@ _JSON_COL_NAMES = {
    "task_payload",
    "history",
 }
+# NOTE: the history tables' JSON payload column is named ``content`` — deliberately
+# NOT added here, because ``mental_models.content`` is plain text (adding "content"
+# would corrupt those reads). The history read paths json.loads ``content`` directly.

 # Columns backed by CLOB in Oracle (large text or JSON). When such a column is
 # returned via a ``RETURNING`` clause it must be bound as DB_TYPE_CLOB; binding
@@ -6249,7 +6249,7 @@ class MemoryEngine(MemoryEngineInterface):
        async with acquire_with_retry(backend) as conn:
            row = await conn.fetchrow(
                f"""
-                SELECT fact_type, history, source_memory_ids
+                SELECT fact_type, source_memory_ids
                FROM {fq_table("memory_units")}
                WHERE id = $1 AND bank_id = $2
                """,
@@ -6261,12 +6261,47 @@ class MemoryEngine(MemoryEngineInterface):
            if row["fact_type"] != "observation":
                return []

-            raw_history = row["history"]
-            if isinstance(raw_history, str):
-                raw_history = json.loads(raw_history)
-            if not raw_history:
+            # History now lives in the dedicated observation_history table
+            # (one row per change), ordered oldest-first to match the prior
+            # append-order semantics the reconstruction below relies on.
+            history_rows = await conn.fetch(
+                f"""
+                SELECT content, changed_at
+                FROM {fq_table("observation_history")}
+                WHERE observation_id = $1
+                ORDER BY changed_at ASC, id ASC
+                """,
+                uuid.UUID(memory_id),
+            )
+            if not history_rows:
                return []

+            def _iso(v: Any) -> Any:
+                return v.isoformat() if hasattr(v, "isoformat") else v
+
+            def _as_list(v: Any) -> list:
+                return list(v) if v else []
+
+            raw_history = []
+            for hr in history_rows:
+                # The snapshot fields live in the JSONB ``content`` blob (str on
+                # Oracle CLOB / when no jsonb codec is registered, dict otherwise).
+                content = hr["content"]
+                if isinstance(content, str):
+                    content = json.loads(content) if content else {}
+                content = content or {}
+                raw_history.append(
+                    {
+                        "previous_text": content.get("previous_text"),
+                        "previous_tags": _as_list(content.get("previous_tags")),
+                        "previous_occurred_start": content.get("previous_occurred_start"),
+                        "previous_occurred_end": content.get("previous_occurred_end"),
+                        "previous_mentioned_at": content.get("previous_mentioned_at"),
+                        "changed_at": _iso(hr["changed_at"]),
+                        "new_source_memory_ids": [str(s) for s in _as_list(content.get("new_source_memory_ids"))],
+                    }
+                )
+
            # Collect all source memory IDs (current full set + all historical new ones)
            current_source_ids: list[str] = [str(sid) for sid in (row["source_memory_ids"] or [])]
            all_source_ids: set[uuid.UUID] = set(uuid.UUID(sid) for sid in current_source_ids)
@@ -8731,7 +8766,7 @@ class MemoryEngine(MemoryEngineInterface):

            rows = await conn.fetch(
                f"""
-                SELECT id, bank_id, text, proof_count, history, tags, source_memory_ids, created_at, updated_at
+                SELECT id, bank_id, text, proof_count, tags, source_memory_ids, created_at, updated_at
                FROM {fq_table("memory_units")}
                WHERE bank_id = $1 AND fact_type = 'observation' {tag_filter}
                ORDER BY updated_at DESC NULLS LAST
@@ -8767,7 +8802,7 @@ class MemoryEngine(MemoryEngineInterface):
        async with acquire_with_retry(backend) as conn:
            row = await conn.fetchrow(
                f"""
-                SELECT id, bank_id, text, proof_count, history, tags, source_memory_ids, created_at, updated_at
+                SELECT id, bank_id, text, proof_count, tags, source_memory_ids, created_at, updated_at
                FROM {fq_table("memory_units")}
                WHERE bank_id = $1 AND id = $2 AND fact_type = 'observation'
                """,
@@ -8808,14 +8843,6 @@ class MemoryEngine(MemoryEngineInterface):

    def _row_to_observation_consolidated(self, row: Any) -> dict[str, Any]:
        """Convert a database row to an observation dict."""
-        import json
-
-        history = row["history"]
-        if isinstance(history, str):
-            history = json.loads(history)
-        elif history is None:
-            history = []
-
        # Convert source_memory_ids to strings
        source_memory_ids = row.get("source_memory_ids") or []
        source_memory_ids = [str(sid) for sid in source_memory_ids]
@@ -8825,7 +8852,8 @@ class MemoryEngine(MemoryEngineInterface):
            "bank_id": row["bank_id"],
            "text": row["text"],
            "proof_count": row["proof_count"] or 1,
-            "history": history,
+            # Deprecated inline field — full history via GET .../{id}/history.
+            "history": [],
            "tags": row["tags"] or [],
            "source_memory_ids": source_memory_ids,
            "source_memories": [],  # Populated separately when fetching full details
@@ -8986,23 +9014,41 @@ class MemoryEngine(MemoryEngineInterface):
        await self._authenticate_tenant(request_context)
        backend = await self._get_backend()
        async with acquire_with_retry(backend) as conn:
-            row = await conn.fetchrow(
-                f"""
-                SELECT history
-                FROM {fq_table("mental_models")}
-                WHERE bank_id = $1 AND id = $2
-                """,
+            exists = await conn.fetchrow(
+                f"SELECT id FROM {fq_table('mental_models')} WHERE bank_id = $1 AND id = $2",
                bank_id,
                mental_model_id,
            )
-            if row is None:
+            if exists is None:
                return None
-            raw_history = row["history"]
-            if isinstance(raw_history, str):
-                raw_history = json.loads(raw_history)
-            if not raw_history:
-                return []
-            return list(reversed(raw_history))
+            # History now lives in the dedicated mental_model_history table (one
+            # row per refresh), returned most-recent-first. The snapshot fields
+            # live in the JSONB ``content`` blob.
+            rows = await conn.fetch(
+                f"""
+                SELECT content, changed_at
+                FROM {fq_table("mental_model_history")}
+                WHERE mental_model_id = $1 AND bank_id = $2
+                ORDER BY changed_at DESC, id DESC
+                """,
+                mental_model_id,
+                bank_id,
+            )
+            result: list[dict] = []
+            for r in rows:
+                content = r["content"]
+                if isinstance(content, str):
+                    content = json.loads(content) if content else {}
+                content = content or {}
+                changed_at = r["changed_at"]
+                result.append(
+                    {
+                        "previous_content": content.get("previous_content"),
+                        "previous_reflect_response": content.get("previous_reflect_response"),
+                        "changed_at": changed_at.isoformat() if hasattr(changed_at, "isoformat") else changed_at,
+                    }
+                )
+            return result

    async def create_mental_model(
        self,
@@ -9521,10 +9567,9 @@ class MemoryEngine(MemoryEngineInterface):
            # If content is changing, fetch current content + reflect_response to record history
            previous_content: str | None = None
            previous_reflect_response: dict[str, Any] | None = None
-            previous_history: list[Any] = []
            if content is not None:
                current_row = await conn.fetchrow(
-                    f"SELECT content, reflect_response, history FROM {fq_table('mental_models')} WHERE bank_id = $1 AND id = $2",
+                    f"SELECT content, reflect_response FROM {fq_table('mental_models')} WHERE bank_id = $1 AND id = $2",
                    bank_id,
                    mental_model_id,
                )
@@ -9535,17 +9580,16 @@ class MemoryEngine(MemoryEngineInterface):
                        previous_reflect_response = json.loads(raw_rr) if raw_rr else None
                    else:
                        previous_reflect_response = raw_rr
-                    raw_history = current_row["history"]
-                    if isinstance(raw_history, str):
-                        raw_history = json.loads(raw_history) if raw_history else []
-                    if isinstance(raw_history, list):
-                        previous_history = raw_history

            # Build dynamic update
            updates = []
            params: list[Any] = [bank_id, mental_model_id]
            param_idx = 3

+            # History snapshot is written to mental_model_history after the UPDATE.
+            record_mm_history = False
+            slim_reflect_response: dict[str, Any] | None = None
+
            if name is not None:
                updates.append(f"name = ${param_idx}")
                params.append(name)
@@ -9556,64 +9600,19 @@ class MemoryEngine(MemoryEngineInterface):
                params.append(content)
                param_idx += 1
                updates.append("last_refreshed_at = NOW()")
-                # Record history entry with the previous content.
-                #
-                # Cap the array to the most recent N entries at write time
-                # (see HINDSIGHT_API_MENTAL_MODEL_HISTORY_MAX_ENTRIES).
-                #
-                # Each entry stores only the slim slice of previous_reflect_response
-                # that consumers actually read: `based_on` (the fact references that
-                # backed that version). The full reflect_response can be hundreds of
-                # KB once `text`, fact bodies, scoring, and embeddings are included;
-                # storing the full payload made each UPDATE rewrite ~10-20 MB of TOAST
-                # per refresh, which prevented HOT updates and accumulated dead
-                # tuples faster than autovacuum could reclaim them. The slim shape
-                # keeps per-row size in the ~hundreds-of-KB range, which fits in a
-                # single heap page and re-enables HOT updates.
+                # Snapshot the previous version for history. The actual write goes
+                # into the dedicated mental_model_history table after the UPDATE
+                # (see _append_mental_model_history); we only store the slim slice
+                # of previous_reflect_response that consumers read — `based_on`,
+                # the fact references that backed that version. The full
+                # reflect_response can be hundreds of KB (text, fact bodies,
+                # scoring, embeddings), so persisting it per entry is wasteful.
                if get_config().enable_mental_model_history:
-                    slim_reflect_response: dict[str, Any] | None = None
                    if previous_reflect_response is not None:
                        based_on = previous_reflect_response.get("based_on")
                        if based_on is not None:
                            slim_reflect_response = {"based_on": based_on}
-                    new_history_entry = {
-                        "previous_content": previous_content,
-                        "previous_reflect_response": slim_reflect_response,
-                        "changed_at": datetime.now(timezone.utc).isoformat(),
-                    }
-                    max_entries = get_config().mental_model_history_max_entries
-                    if self._database_backend_type == "oracle":
-                        # Oracle has no jsonb_agg / jsonb_array_elements WITH ORDINALITY,
-                        # so the PG read-modify-write below cannot be rewritten. Compute the
-                        # trimmed history in Python (we already fetched the current array) and
-                        # bind it as a single JSON value. Trimming keeps the most recent
-                        # ``max_entries`` entries — identical semantics to the PG GREATEST/idx
-                        # window: keep [] when max_entries == 0, else the last max_entries.
-                        combined = previous_history + [new_history_entry]
-                        trimmed = [] if max_entries == 0 else combined[-max_entries:]
-                        updates.append(f"history = ${param_idx}")
-                        params.append(json.dumps(trimmed))
-                        param_idx += 1
-                    else:
-                        history_entry = json.dumps([new_history_entry])
-                        history_param_idx = param_idx
-                        param_idx += 1
-                        max_entries_param_idx = param_idx
-                        param_idx += 1
-                        updates.append(
-                            "history = ("
-                            "  SELECT COALESCE(jsonb_agg(elem ORDER BY idx), '[]'::jsonb) "
-                            "  FROM jsonb_array_elements("
-                            f"    COALESCE(history, '[]'::jsonb) || ${history_param_idx}::jsonb"
-                            "  ) WITH ORDINALITY a(elem, idx) "
-                            "  WHERE idx > GREATEST("
-                            "    jsonb_array_length(COALESCE(history, '[]'::jsonb)) + 1"
-                            f"    - ${max_entries_param_idx}, 0"
-                            "  )"
-                            ")"
-                        )
-                        params.append(history_entry)
-                        params.append(max_entries)
+                    record_mm_history = True
                # Also update embedding (convert to string for asyncpg vector type)
                embedding_text = f"{name or ''} {content}"
                embedding = await embedding_utils.generate_embeddings_batch(self.embeddings, [embedding_text])
@@ -9671,8 +9670,64 @@ class MemoryEngine(MemoryEngineInterface):

            row = await conn.fetchrow(query, *params)

+            # Persist the previous-version snapshot in the dedicated history table
+            # (one row per refresh), then trim to the configured cap. Replaces the
+            # old single-JSONB-column append, which rewrote the whole array (plus
+            # TOAST) on every refresh and was capped by entry count, not size.
+            if row is not None and record_mm_history:
+                await self._append_mental_model_history(
+                    conn,
+                    bank_id,
+                    mental_model_id,
+                    previous_content,
+                    slim_reflect_response,
+                    get_config().mental_model_history_max_entries,
+                )
+
            return self._row_to_mental_model(row) if row else None

+    async def _append_mental_model_history(
+        self,
+        conn: Any,
+        bank_id: str,
+        mental_model_id: str,
+        previous_content: str | None,
+        previous_reflect_response: dict[str, Any] | None,
+        max_entries: int,
+    ) -> None:
+        """Insert one refresh snapshot into mental_model_history, then delete the
+        oldest rows beyond ``max_entries`` for this model. The snapshot is stored
+        as a single JSONB ``content`` blob (per-row, so it stays small); bounding
+        by row count keeps per-model history from growing without bound."""
+        content = json.dumps(
+            {"previous_content": previous_content, "previous_reflect_response": previous_reflect_response}
+        )
+        await conn.execute(
+            f"""
+            INSERT INTO {fq_table("mental_model_history")} (mental_model_id, bank_id, content, changed_at)
+            VALUES ($1, $2, $3::jsonb, now())
+            """,
+            mental_model_id,
+            bank_id,
+            content,
+        )
+        if max_entries and max_entries > 0:
+            await conn.execute(
+                f"""
+                DELETE FROM {fq_table("mental_model_history")}
+                WHERE mental_model_id = $1 AND bank_id = $2
+                  AND id NOT IN (
+                      SELECT id FROM {fq_table("mental_model_history")}
+                      WHERE mental_model_id = $1 AND bank_id = $2
+                      ORDER BY changed_at DESC, id DESC
+                      LIMIT $3
+                  )
+                """,
+                mental_model_id,
+                bank_id,
+                max_entries,
+            )
+
    async def clear_mental_model(
        self,
        bank_id: str,
@@ -56,11 +56,22 @@ _REPLAYED_TABLES = frozenset(
        "unit_entities",
        "memory_links",
        "entity_cooccurrences",
+        # observation_history FKs to a memory_units observation, but observations
+        # are derived: they're regenerated with FRESH ids when consolidation is
+        # replayed on import (see _EXPORTED_FACT_TYPES — observations are excluded).
+        # There is no stable observation id to re-attach history to, so it is not
+        # carried; the target rebuilds observation history as it re-consolidates.
+        "observation_history",
    }
 )
 # Carried verbatim as JSON rows (bank config + synthesized state). Embedding-bearing
 # rows have their vector stripped (see _DERIVED_COLUMNS) and are re-embedded on import.
 _BANK_ROW_TABLES = ("banks", "mental_models", "directives", "webhooks")
+# Bank-scoped child-history carried verbatim. Unlike observations, mental models
+# keep their (id, bank_id) across export/import, so their refresh history can be
+# re-attached. The surrogate ``id`` is dropped on dump so the target reassigns it
+# (see _dump_history_rows); restored after its parent table (mental_models).
+_CARRIED_HISTORY_TABLES = ("mental_model_history",)
 # Operational history — only carried with include_history=True.
 _HISTORY_TABLES = ("audit_log", "llm_requests")
 # Intentionally never exported.
@@ -234,6 +245,21 @@ async def _dump_bank_rows(conn: Any, table: str, bank_id: str) -> list[dict]:
    return [{k: v for k, v in dict(row).items() if k not in _DERIVED_COLUMNS} for row in rows]


+async def _dump_history_rows(conn: Any, table: str, bank_id: str) -> list[dict]:
+    """Dump a bank-scoped child-history table for carrying across instances.
+
+    Drops the surrogate ``id`` so the target reassigns it from its own IDENTITY
+    sequence (carrying explicit ids would leave the sequence un-advanced and
+    collide with later writes). Ordered oldest-first so the reassigned ids keep
+    the same chronological tie-break order the read path relies on.
+    """
+    rows = await conn.fetch(
+        f"SELECT * FROM {fq_table(table)} WHERE bank_id = $1 ORDER BY changed_at, id",
+        bank_id,
+    )
+    return [{k: v for k, v in dict(row).items() if k not in _DERIVED_COLUMNS and k != "id"} for row in rows]
+
+
 async def export_bank(conn: Any, bank_id: str, *, include_history: bool = False) -> bytes:
    """Export an entire bank into a portable ZIP archive (no embeddings).

@@ -255,6 +281,8 @@ async def export_bank(conn: Any, bank_id: str, *, include_history: bool = False)
    observations = await _load_observations(conn, bank_id, loaded.unit_index)

    bank_rows = {table: await _dump_bank_rows(conn, table, bank_id) for table in _BANK_ROW_TABLES}
+    for table in _CARRIED_HISTORY_TABLES:
+        bank_rows[table] = await _dump_history_rows(conn, table, bank_id)
    history_rows: dict[str, list[dict]] = {}
    if include_history:
        history_rows = {table: await _dump_bank_rows(conn, table, bank_id) for table in _HISTORY_TABLES}
@@ -218,6 +218,10 @@ async def import_documents(
 # Bank-level config/state tables restored verbatim from a whole-bank archive.
 # Order matters for foreign keys: banks (parent) is restored before any child.
 _BANK_CHILD_TABLES = ("mental_models", "directives", "webhooks")
+# Child-history carried verbatim; restored after its parent (mental_models) so the
+# foreign key resolves. Surrogate ids were dropped on export (the target reassigns
+# them), so these restore via fresh IDENTITY values.
+_CARRIED_HISTORY_TABLES = ("mental_model_history",)
 _HISTORY_TABLES = ("audit_log", "llm_requests")


@@ -230,6 +234,7 @@ class BankImportResult:
    facts_imported: int = 0
    observations_imported: int = 0
    mental_models_imported: int = 0
+    mental_model_history_imported: int = 0
    directives_imported: int = 0
    webhooks_imported: int = 0
    history_rows_imported: int = 0
@@ -258,7 +263,7 @@ def parse_bank_archive(archive_bytes: bytes) -> ParsedBankArchive:
                f"Not a whole-bank archive (archive_type={manifest.archive_type!r}); use import_documents instead"
            )
        bank_rows: dict[str, list[dict]] = {}
-        for table in ("banks", *_BANK_CHILD_TABLES):
+        for table in ("banks", *_BANK_CHILD_TABLES, *_CARRIED_HISTORY_TABLES):
            fname = f"{table}.json"
            bank_rows[table] = json.loads(zf.read(fname)) if fname in names else []
        history_rows: dict[str, list[dict]] = {}
@@ -396,6 +401,10 @@ async def import_bank(
        result.mental_models_imported = await _restore_rows(
            conn, "mental_models", parsed.bank_rows.get("mental_models", [])
        )
+        # Restored after mental_models so the (mental_model_id, bank_id) FK resolves.
+        result.mental_model_history_imported = await _restore_rows(
+            conn, "mental_model_history", parsed.bank_rows.get("mental_model_history", [])
+        )
        result.directives_imported = await _restore_rows(conn, "directives", parsed.bank_rows.get("directives", []))
        result.webhooks_imported = await _restore_rows(conn, "webhooks", parsed.bank_rows.get("webhooks", []))
        if include_history:
@@ -404,12 +413,13 @@ async def import_bank(

    logger.info(
        "[transfer] Imported bank %s: %d doc(s), %d fact(s), %d observation(s), "
-        "%d mental model(s), %d directive(s), %d webhook(s), %d history row(s)",
+        "%d mental model(s), %d mm-history row(s), %d directive(s), %d webhook(s), %d history row(s)",
        bank_id,
        result.documents_imported,
        result.facts_imported,
        result.observations_imported,
        result.mental_models_imported,
+        result.mental_model_history_imported,
        result.directives_imported,
        result.webhooks_imported,
        result.history_rows_imported,
@@ -464,7 +464,7 @@ class TestConsolidationIntegration:
        async with memory._pool.acquire() as conn:
            observations = await conn.fetch(
                """
-                SELECT id, text, source_memory_ids, history
+                SELECT id, text, source_memory_ids
                FROM memory_units
                WHERE bank_id = $1 AND fact_type = 'observation'
                """,
@@ -99,12 +99,19 @@ def test_export_bank_covers_schema():
    from hindsight_api.admin.cli import BACKUP_TABLES
    from hindsight_api.engine.transfer.export import (
        _BANK_ROW_TABLES,
+        _CARRIED_HISTORY_TABLES,
        _HISTORY_TABLES,
        _REPLAYED_TABLES,
        _SKIP_TABLES,
    )

-    buckets = [set(_REPLAYED_TABLES), set(_BANK_ROW_TABLES), set(_HISTORY_TABLES), set(_SKIP_TABLES)]
+    buckets = [
+        set(_REPLAYED_TABLES),
+        set(_BANK_ROW_TABLES),
+        set(_CARRIED_HISTORY_TABLES),
+        set(_HISTORY_TABLES),
+        set(_SKIP_TABLES),
+    ]
    classified = set().union(*buckets)
    assert classified == set(BACKUP_TABLES), (
        f"export-bank classification drifted from BACKUP_TABLES: "
@@ -149,6 +156,7 @@ async def test_export_bank_contents(memory, request_context):
        assert manifest.document_count == 1
        assert manifest.webhook_count == 1
        assert "mental_models.json" in names and "directives.json" in names
+        assert "mental_model_history.json" in names
        assert any(d.endswith(".json") and d.startswith("documents/") for d in names)
        # No history files unless requested.
        assert not any(n.startswith("history/") for n in names)
@@ -316,6 +324,47 @@ async def test_bank_export_import_exact_roundtrip(memory, request_context):
        await memory.delete_bank(bank, request_context=request_context)


+@pytest.mark.asyncio
+async def test_bank_roundtrip_carries_mental_model_history(memory, request_context):
+    """Mental-model refresh history survives export/import. Mental models keep a
+    stable (id, bank_id), so the dedicated mental_model_history rows are carried
+    (the surrogate id is dropped on export; the target reassigns it)."""
+    bank = _unique_bank("bank_mm_hist")
+    try:
+        await memory.get_bank_profile(bank, request_context=request_context)
+        await memory.create_mental_model(
+            bank,
+            name="Work model",
+            source_query="where do people work",
+            content="v1",
+            mental_model_id="mm-1",
+            request_context=request_context,
+        )
+        await memory.update_mental_model(
+            bank, mental_model_id="mm-1", content="v2", request_context=request_context
+        )
+        await memory.update_mental_model(
+            bank, mental_model_id="mm-1", content="v3", request_context=request_context
+        )
+        # Two refreshes → two snapshots (previous content v1 then v2), newest-first.
+        before = await memory.get_mental_model_history(bank, "mm-1", request_context=request_context)
+        assert [h["previous_content"] for h in before] == ["v2", "v1"]
+
+        from hindsight_api.engine.transfer import export_bank
+
+        backend = await memory._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            archive = await export_bank(conn, bank)
+        await memory.delete_bank(bank, request_context=request_context)
+        result = await memory.import_bank_async(archive, request_context)
+        assert result.mental_model_history_imported == 2
+
+        after = await memory.get_mental_model_history(bank, "mm-1", request_context=request_context)
+        assert [h["previous_content"] for h in after] == ["v2", "v1"]
+    finally:
+        await memory.delete_bank(bank, request_context=request_context)
+
+
@pytest.mark.asyncio
 async def test_import_bank_rejects_documents_archive(memory, request_context):
    """A documents-only archive must be rejected by the bank importer."""
@@ -186,8 +186,8 @@ async def test_graph_document_filter_includes_observations_via_source_memories(
        # World fact tied to the document.
        await conn.execute(
            """
-            INSERT INTO memory_units (id, bank_id, text, fact_type, document_id, history)
-            VALUES ($1, $2, $3, 'world', $4, '[]'::jsonb)
+            INSERT INTO memory_units (id, bank_id, text, fact_type, document_id)
+            VALUES ($1, $2, $3, 'world', $4)
            """,
            fact_id,
            bank_id,
@@ -198,8 +198,8 @@ async def test_graph_document_filter_includes_observations_via_source_memories(
        # Unrelated fact NOT tied to the document.
        await conn.execute(
            """
-            INSERT INTO memory_units (id, bank_id, text, fact_type, history)
-            VALUES ($1, $2, $3, 'world', '[]'::jsonb)
+            INSERT INTO memory_units (id, bank_id, text, fact_type)
+            VALUES ($1, $2, $3, 'world')
            """,
            other_fact_id,
            bank_id,
@@ -210,9 +210,9 @@ async def test_graph_document_filter_includes_observations_via_source_memories(
        await conn.execute(
            """
            INSERT INTO memory_units (
-                id, bank_id, text, fact_type, source_memory_ids, history, proof_count
+                id, bank_id, text, fact_type, source_memory_ids, proof_count
            )
-            VALUES ($1, $2, $3, 'observation', $4::uuid[], '[]'::jsonb, 1)
+            VALUES ($1, $2, $3, 'observation', $4::uuid[], 1)
            """,
            observation_id,
            bank_id,
@@ -224,9 +224,9 @@ async def test_graph_document_filter_includes_observations_via_source_memories(
        await conn.execute(
            """
            INSERT INTO memory_units (
-                id, bank_id, text, fact_type, source_memory_ids, history, proof_count
+                id, bank_id, text, fact_type, source_memory_ids, proof_count
            )
-            VALUES ($1, $2, $3, 'observation', $4::uuid[], '[]'::jsonb, 1)
+            VALUES ($1, $2, $3, 'observation', $4::uuid[], 1)
            """,
            unrelated_observation_id,
            bank_id,
@@ -231,7 +231,13 @@ async def test_horse_farm_observation_history(memory_real_llm: MemoryEngine, req
            async with pool.acquire() as conn:
                observations = await conn.fetch(
                    """
-                    SELECT id, text, proof_count, source_memory_ids, history
+                    SELECT id, text, proof_count, source_memory_ids,
+                           COALESCE((
+                               SELECT jsonb_agg(jsonb_build_object('previous_text', oh.content->>'previous_text')
+                                                ORDER BY oh.changed_at, oh.id)
+                               FROM observation_history oh
+                               WHERE oh.observation_id = memory_units.id
+                           ), '[]'::jsonb) AS history
                    FROM memory_units
                    WHERE bank_id = $1 AND fact_type = 'observation'
                    ORDER BY created_at
@@ -255,7 +261,13 @@ async def test_horse_farm_observation_history(memory_real_llm: MemoryEngine, req
    async with pool.acquire() as conn:
        observations = await conn.fetch(
            """
-            SELECT id, text, proof_count, source_memory_ids, history
+            SELECT id, text, proof_count, source_memory_ids,
+                   COALESCE((
+                       SELECT jsonb_agg(jsonb_build_object('previous_text', oh.content->>'previous_text')
+                                        ORDER BY oh.changed_at, oh.id)
+                       FROM observation_history oh
+                       WHERE oh.observation_id = memory_units.id
+                   ), '[]'::jsonb) AS history
            FROM memory_units
            WHERE bank_id = $1 AND fact_type = 'observation'
            ORDER BY created_at
@@ -0,0 +1,78 @@
+"""Deterministic tests for the dedicated observation_history table.
+
+The history of an observation lives in its own table (one row per change),
+written by consolidator._append_observation_history and read back by
+MemoryEngine.get_observation_history. These tests exercise that path directly
+(no LLM) so the array/timestamptz binding and the per-observation cap are
+covered without depending on consolidation deciding to issue an UPDATE.
+"""
+
+import uuid
+from typing import Any
+
+import pytest
+
+from hindsight_api.config import clear_config_cache
+from hindsight_api.engine.consolidation import consolidator as consolidator_mod
+from hindsight_api.engine.consolidation.consolidator import _ObservationHistorySnapshot
+from hindsight_api.engine.memory_engine import MemoryEngine
+
+
+def _entry(i: int) -> _ObservationHistorySnapshot:
+    return _ObservationHistorySnapshot(
+        previous_text=f"v{i}",
+        previous_tags=[f"tag{i}"],
+        previous_occurred_start=None,
+        previous_occurred_end=None,
+        previous_mentioned_at="2025-01-01T00:00:00Z",
+        new_source_memory_ids=[],
+    )
+
+
+@pytest.mark.asyncio
+class TestObservationHistory:
+    async def test_append_read_and_cap(
+        self, memory: MemoryEngine, request_context: Any, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Snapshots are appended as rows, returned oldest-first, and trimmed to
+        the configured cap (the oldest over-cap rows are deleted on each write)."""
+        monkeypatch.setenv("HINDSIGHT_API_OBSERVATION_HISTORY_MAX_ENTRIES", "3")
+        clear_config_cache()
+
+        bank_id = f"test-obs-hist-{uuid.uuid4().hex[:8]}"
+        await memory.get_bank_profile(bank_id, request_context=request_context)
+
+        obs_id = uuid.uuid4()
+        pool = await memory._get_pool()
+        async with pool.acquire() as conn:
+            await conn.execute(
+                """
+                INSERT INTO memory_units (id, bank_id, text, event_date, fact_type)
+                VALUES ($1, $2, 'current observation', now(), 'observation')
+                """,
+                obs_id,
+                bank_id,
+            )
+            # 5 updates → 5 snapshots → trimmed to the most recent 3.
+            for i in range(1, 6):
+                await consolidator_mod._append_observation_history(conn, bank_id, str(obs_id), _entry(i), 3)
+
+        history = await memory.get_observation_history(bank_id, str(obs_id), request_context=request_context)
+
+        # Capped to the most-recent 3 (v3, v4, v5), returned oldest-first.
+        assert [h["previous_text"] for h in history] == ["v3", "v4", "v5"]
+        # Array column round-trips (the binding the mental-model path doesn't exercise).
+        assert history[0]["previous_tags"] == ["tag3"]
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+    async def test_returns_none_for_missing_observation(
+        self, memory: MemoryEngine, request_context: Any
+    ) -> None:
+        bank_id = f"test-obs-hist-{uuid.uuid4().hex[:8]}"
+        await memory.get_bank_profile(bank_id, request_context=request_context)
+        result = await memory.get_observation_history(
+            bank_id, str(uuid.uuid4()), request_context=request_context
+        )
+        assert result is None
+        await memory.delete_bank(bank_id, request_context=request_context)
@@ -117,9 +117,9 @@ async def seeded(memory_no_llm_verify: MemoryEngine):
            """
            INSERT INTO memory_units (
                id, bank_id, text, fact_type, embedding, event_date,
-                source_memory_ids, history, proof_count
+                source_memory_ids, proof_count
            )
-            VALUES ($1, $2, $3, 'observation', $4::vector, now(), $5::uuid[], '[]'::jsonb, 1)
+            VALUES ($1, $2, $3, 'observation', $4::vector, now(), $5::uuid[], 1)
            """,
            ID_OBS_INHERITED,
            bank_id,
@@ -133,9 +133,9 @@ async def seeded(memory_no_llm_verify: MemoryEngine):
            """
            INSERT INTO memory_units (
                id, bank_id, text, fact_type, embedding, event_date,
-                source_memory_ids, history, proof_count
+                source_memory_ids, proof_count
            )
-            VALUES ($1, $2, $3, 'observation', $4::vector, now(), NULL, '[]'::jsonb, 1)
+            VALUES ($1, $2, $3, 'observation', $4::vector, now(), NULL, 1)
            """,
            ID_OBS_DIRECT,
            bank_id,
@@ -624,11 +624,11 @@ async def _insert_synthetic_observations(pool: Any, bank_id: str) -> int:
            f"""
            INSERT INTO {table} (
                id, bank_id, text, fact_type, embedding,
-                proof_count, source_memory_ids, history,
+                proof_count, source_memory_ids,
                tags, event_date, occurred_start, occurred_end, mentioned_at
            ) VALUES (
                $1, $2, $3, 'observation', $4::vector,
-                1, ARRAY[$5::uuid], '[]'::jsonb,
+                1, ARRAY[$5::uuid],
                $6, $7, $8, $9, $10
            )
            ON CONFLICT DO NOTHING
@@ -959,8 +959,8 @@ For advanced authentication (JWT, OAuth, multi-tenant schemas), implement a cust
 | `HINDSIGHT_API_RECALL_MAX_CANDIDATES_PER_SOURCE` | Cap on candidates each retrieval source (semantic, BM25, graph, temporal) contributes to RRF, applied before the global reranker cap. Prevents one over-expanding backend from filling the reranker budget on its own. `0` disables the cap. | `0` |
 | `HINDSIGHT_API_RECALL_STRATEGY_BOOSTS` | Prioritise one or more retrieval sources over the others on recall, as a comma-separated `strategy:level` list (e.g. `graph:high` to strongly favour graph hits, or `graph:high,bm25:low`). Strategies: `semantic`, `bm25`, `graph`, `temporal`. Levels: `low` (gentle — mainly protects the source's candidates from being dropped before reranking), `medium` (moderate preference), `high` (strong — the source dominates the candidate pool and outranks most other matches, only a strong direct match still wins). The boost is applied in two places: before the reranker cap (so favoured candidates survive the `HINDSIGHT_API_RERANKER_MAX_CANDIDATES` budget) and after reranking (to nudge them up the final order); a named level is used because those two stages live on different score scales. Only the strategies you list are boosted — any you omit keep their normal weight (no implicit boost). A strategy written without a level (`graph` or `graph:`) defaults to `medium`. Empty disables the feature. | _(empty)_ |
 | `HINDSIGHT_API_MENTAL_MODEL_REFRESH_CONCURRENCY` | Max concurrent mental model refreshes | `8` |
-| `HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY` | Track history of content changes to each mental model (previous content + timestamp). Disable to reduce storage if audit trails are not needed. | `true` |
-| `HINDSIGHT_API_MENTAL_MODEL_HISTORY_MAX_ENTRIES` | Max entries retained in the per-mental-model history jsonb array. Older entries are dropped at write time. Prevents the array from crossing Postgres's hard 256MB jsonb size limit (which would otherwise make further UPDATEs to the row fail with SQLSTATE 54000). Each entry stores only the slim `{based_on}` slice of the prior `reflect_response` (the only field consumed by the control-plane UI's history view) so per-row size stays bounded and HOT updates apply. | `50` |
+| `HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY` | Track history of content changes to each mental model (previous content + timestamp), stored one row per change in the `mental_model_history` table. Set to `false` to disable entirely — no history rows are written, reducing storage if audit trails are not needed. **This is how you turn the feature off** (not a zero cap). | `true` |
+| `HINDSIGHT_API_MENTAL_MODEL_HISTORY_MAX_ENTRIES` | Max history rows kept per mental model. On each refresh the previous version is inserted into the `mental_model_history` table and the oldest rows beyond this cap are deleted, so per-model history can't grow without bound. `0` or a negative value **removes the cap** (history then grows with every refresh — unbounded); to turn history off entirely set `HINDSIGHT_API_ENABLE_MENTAL_MODEL_HISTORY=false` instead. | `50` |

 #### Graph Retrieval Algorithm

@@ -1327,7 +1327,8 @@ Observations are deduplicated, evidence-grounded knowledge consolidated from mul
 |----------|-------------|---------|
 | `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `true` |
 | `HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION` | Automatically trigger consolidation after retain, delete, and update operations. When `false`, consolidation only runs when explicitly triggered via the [consolidate endpoint](/developer/api/operations#consolidation). Configurable per bank. | `true` |
-| `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY` | Track history of changes to each observation (previous content + timestamp). Disable to reduce storage if audit trails are not needed. | `true` |
+| `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY` | Track history of changes to each observation (previous text/tags/dates + timestamp), stored one row per change in the `observation_history` table. Set to `false` to disable entirely — no history rows are written. **This is how you turn the feature off** (not a zero cap). | `true` |
+| `HINDSIGHT_API_OBSERVATION_HISTORY_MAX_ENTRIES` | Max history rows kept per observation. On each update the previous version is inserted into the `observation_history` table and the oldest rows beyond this cap are deleted, so an often-reinforced observation's history can't grow without bound. `0` or a negative value **removes the cap** (unbounded); to turn history off entirely set `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY=false` instead. | `50` |
 | `HINDSIGHT_API_CONSOLIDATION_MAX_ATTEMPTS` | Outer retry attempts for the consolidation LLM batch call. Each attempt uses the inner retry budget (`HINDSIGHT_API_CONSOLIDATION_LLM_MAX_RETRIES`). Worst-case API calls per batch = `MAX_ATTEMPTS × (LLM_MAX_RETRIES + 1)`. | `3` |
 | `HINDSIGHT_API_CONSOLIDATION_BATCH_SIZE` | Memories to load per batch (internal optimization) | `50` |
 | `HINDSIGHT_API_CONSOLIDATION_MAX_MEMORIES_PER_ROUND` | Maximum memories processed per consolidation round. When the limit is reached, the job yields its worker slot and re-queues itself so other banks get fair scheduling. Mental model refreshes only run on the final round. `0` = unlimited. Configurable per bank. | `100` |