fix(transfer): preserve JSONB and timestamp provenance

hindsight-api-slim/hindsight_api/admin/cli.py

@@ -364,7 +364,14 @@ async def _run_export_bank(db_url: str, bank_id: str, output: Path, schema: str,
         # export_bank resolves table names via fq_table (the _current_schema
         # contextvar); set it so the raw connection targets the right schema.
         _current_schema.set(schema)
-        data = await export_bank(conn, bank_id, include_history=include_history)
+        # _admin_connect registers JSON codecs, so row dumps already contain
+        # decoded Python values (including JSON scalar strings).
+        data = await export_bank(
+            conn,
+            bank_id,
+            include_history=include_history,
+            bank_rows_json_encoding="decoded",
+        )
     finally:
         await conn.close()
 

hindsight-api-slim/hindsight_api/engine/transfer/export.py

@@ -26,6 +26,7 @@
     CARRIED_HISTORY_TABLES,
     HISTORY_TABLES,
     SCHEMA_VERSION,
+    BankRowsJSONEncoding,
     TransferCausalRelation,
     TransferChunk,
     TransferDocument,
@@ -138,7 +139,12 @@ def _as_jsonb(value: Any) -> Any:
     if value is None:
         return None
     if isinstance(value, str):
-        return json.loads(value)
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError:
+            # Admin connections register a JSONB decoder, so a valid scalar such
+            # as `"combined"` arrives here as the already-decoded `combined`.
+            return value
     return value
 
 
@@ -266,7 +272,13 @@ async def _dump_history_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 and k != "id"} for row in rows]
 
 
-async def export_bank(conn: Any, bank_id: str, *, include_history: bool = False) -> bytes:
+async def export_bank(
+    conn: Any,
+    bank_id: str,
+    *,
+    include_history: bool = False,
+    bank_rows_json_encoding: BankRowsJSONEncoding = "serialized",
+) -> bytes:
     """Export an entire bank into a portable ZIP archive (no embeddings).
 
     Produces a superset of the documents archive: the logical
@@ -321,6 +333,7 @@ async def export_bank(conn: Any, bank_id: str, *, include_history: bool = False)
             directive_count=len(bank_rows.get("directives", [])),
             webhook_count=len(bank_rows.get("webhooks", [])),
             includes_history=include_history,
+            bank_rows_json_encoding=bank_rows_json_encoding,
         )
         zf.writestr("manifest.json", manifest.model_dump_json(indent=2))
 

hindsight-api-slim/hindsight_api/engine/transfer/importer.py

@@ -33,6 +33,7 @@
     CARRIED_HISTORY_TABLES,
     HISTORY_TABLES,
     SCHEMA_VERSION,
+    BankRowsJSONEncoding,
     TransferDocument,
     TransferFact,
     TransferManifest,
@@ -275,7 +276,18 @@ def parse_bank_archive(archive_bytes: bytes) -> ParsedBankArchive:
     return ParsedBankArchive(manifest=manifest, bank_rows=bank_rows, history_rows=history_rows)
 
 
-async def _restore_rows(conn: Any, table: str, rows: list[dict]) -> int:
+def _resolve_bank_rows_json_encoding(manifest: TransferManifest) -> BankRowsJSONEncoding:
+    """Resolve row JSON provenance, including the released v1 archive contract."""
+    return manifest.bank_rows_json_encoding or "decoded"
+
+
+async def _restore_rows(
+    conn: Any,
+    table: str,
+    rows: list[dict],
+    *,
+    bank_rows_json_encoding: BankRowsJSONEncoding = "decoded",
+) -> int:
     """Insert verbatim rows into a bank-scoped table, coercing JSON-encoded values
     back to the column's type (timestamps, uuids, jsonb). ``ON CONFLICT DO NOTHING``
     keeps an import idempotent and safe to re-run against a partially-filled target."""
@@ -302,9 +314,12 @@ async def _restore_rows(conn: Any, table: str, rows: list[dict]) -> int:
             value = row[col]
             if data_type in ("jsonb", "json"):
                 # asyncpg has no JSON codec on these raw connections; pass JSON
-                # text and cast. Values may already be str (no codec on export) or
-                # a Python object (codec on export) — normalize to text either way.
-                values.append(value if isinstance(value, str) or value is None else json.dumps(value))
+                # text and cast. Provenance is required because a decoded JSON
+                # scalar containing JSON text is indistinguishable from a raw
+                # serialized object after the outer archive JSON is parsed.
+                if value is not None and (bank_rows_json_encoding == "decoded" or not isinstance(value, str)):
+                    value = json.dumps(value)
+                values.append(value)
                 placeholders.append(f"${position}::jsonb")
                 continue
             if value is not None and isinstance(value, str):
@@ -353,6 +368,7 @@ async def import_bank(
     if ops is None:
         ops = backend.ops
     parsed = parse_bank_archive(archive_bytes)
+    bank_rows_json_encoding = _resolve_bank_rows_json_encoding(parsed.manifest)
     source_bank_id = parsed.manifest.source_bank_id
     bank_id = target_bank_id or source_bank_id
 
@@ -375,7 +391,12 @@ async def import_bank(
                 f"(it is not a merge). Delete the bank first, or pass a different target bank id."
             )
         # Bank row first — children (documents, mental_models, …) FK to it.
-        await _restore_rows(conn, "banks", parsed.bank_rows.get("banks", []))
+        await _restore_rows(
+            conn,
+            "banks",
+            parsed.bank_rows.get("banks", []),
+            bank_rows_json_encoding=bank_rows_json_encoding,
+        )
     # Ensure the bank's per-bank vector indexes exist (no-op for global-index
     # extensions); idempotent and keeps the restored banks row (ON CONFLICT DO NOTHING).
     await bank_utils.get_or_create_bank_profile(backend, bank_id)
@@ -400,17 +421,38 @@ async def import_bank(
     )
     async with acquire_with_retry(backend) as conn:
         result.mental_models_imported = await _restore_rows(
-            conn, "mental_models", parsed.bank_rows.get("mental_models", [])
+            conn,
+            "mental_models",
+            parsed.bank_rows.get("mental_models", []),
+            bank_rows_json_encoding=bank_rows_json_encoding,
         )
         # 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", [])
+            conn,
+            "mental_model_history",
+            parsed.bank_rows.get("mental_model_history", []),
+            bank_rows_json_encoding=bank_rows_json_encoding,
+        )
+        result.directives_imported = await _restore_rows(
+            conn,
+            "directives",
+            parsed.bank_rows.get("directives", []),
+            bank_rows_json_encoding=bank_rows_json_encoding,
+        )
+        result.webhooks_imported = await _restore_rows(
+            conn,
+            "webhooks",
+            parsed.bank_rows.get("webhooks", []),
+            bank_rows_json_encoding=bank_rows_json_encoding,
         )
-        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:
             for table in HISTORY_TABLES:
-                result.history_rows_imported += await _restore_rows(conn, table, parsed.history_rows.get(table, []))
+                result.history_rows_imported += await _restore_rows(
+                    conn,
+                    table,
+                    parsed.history_rows.get(table, []),
+                    bank_rows_json_encoding=bank_rows_json_encoding,
+                )
 
     logger.info(
         "[transfer] Imported bank %s: %d doc(s), %d fact(s), %d observation(s), "
@@ -521,6 +563,15 @@ async def _import_one_document(
                 document.tags,
                 ops=ops,
             )
+            if document.created_at is not None:
+                # Transfer archives carry source provenance. Apply it here,
+                # without changing normal retain/upsert timestamp semantics.
+                await conn.execute(
+                    f"UPDATE {fq_table('documents')} SET created_at = $1 WHERE id = $2 AND bank_id = $3",
+                    document.created_at,
+                    target_id,
+                    bank_id,
+                )
 
             chunk_id_map: dict[int, str] = {}
             if chunk_meta:
@@ -640,11 +691,19 @@ async def _import_observations(
 
             all_source_ids: set[uuid.UUID] = set()
             for (obs, sources), obs_unit_id in zip(resolved, obs_unit_ids):
+                observation_uuid = uuid.UUID(obs_unit_id)
+                if obs.event_date is not None:
+                    # insert_facts_batch derives event_date for normal writes;
+                    # transfer restores the source value carried by the archive.
+                    await conn.execute(
+                        f"UPDATE {fq_table('memory_units')} SET event_date = $1 WHERE id = $2 AND bank_id = $3",
+                        obs.event_date,
+                        observation_uuid,
+                        bank_id,
+                    )
                 source_uuids = [uuid.UUID(s) for s in sources]
                 all_source_ids.update(source_uuids)
-                await _link_observation_sources(
-                    conn, ops, bank_id, uuid.UUID(obs_unit_id), source_uuids, obs.proof_count
-                )
+                await _link_observation_sources(conn, ops, bank_id, observation_uuid, source_uuids, obs.proof_count)
 
             # Mark source facts consolidated so the target consolidator skips them.
             if all_source_ids:

hindsight-api-slim/hindsight_api/engine/transfer/schema.py

@@ -29,6 +29,7 @@
 HISTORY_TABLES = ("audit_log", "llm_requests")
 
 ObservationScopes = Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]]
+BankRowsJSONEncoding = Literal["decoded", "serialized"]
 
 
 class TransferCausalRelation(BaseModel):
@@ -142,3 +143,7 @@ class TransferManifest(BaseModel):
     webhook_count: int = 0
     # True when --include-history carried audit_log / llm_requests.
     includes_history: bool = False
+    # How JSON/JSONB values in bank/history row files were represented by the
+    # producing connection. Absent on legacy v1 archives; import treats those as
+    # decoded because the released producer was the codec-enabled admin CLI.
+    bank_rows_json_encoding: BankRowsJSONEncoding | None = None

hindsight-api-slim/tests/test_admin_bank_transfer.py

@@ -0,0 +1,49 @@
+"""Tests for the admin CLI whole-bank transfer boundary."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
+
+from hindsight_api.admin import cli
+
+
+class _FakeConnection:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+@pytest.mark.asyncio
+async def test_run_export_bank_declares_decoded_json_rows(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """The codec-enabled admin producer must identify its rows as decoded."""
+    connection = _FakeConnection()
+    export_bank = AsyncMock(return_value=b"archive")
+
+    async def fake_admin_connect(db_url: str) -> _FakeConnection:
+        assert db_url == "postgresql://example"
+        return connection
+
+    monkeypatch.setattr(cli, "_admin_connect", fake_admin_connect)
+    monkeypatch.setattr(cli, "export_bank", export_bank)
+
+    output = tmp_path / "bank.zip"
+    size = await cli._run_export_bank(
+        "postgresql://example",
+        "source-bank",
+        output,
+        "tenant_schema",
+        include_history=True,
+    )
+
+    export_bank.assert_awaited_once_with(
+        connection,
+        "source-bank",
+        include_history=True,
+        bank_rows_json_encoding="decoded",
+    )
+    assert output.read_bytes() == b"archive"
+    assert size == len(b"archive")
+    assert connection.closed is True