fix(transfer): preserve archive provenance

Record bank-row JSON encoding in transfer manifests so decoded scalar strings and serialized objects restore without ambiguous parsing. Preserve archived document and observation timestamps during replay.

Co-Authored-By: OpenAI GPT-5 Codex medium <noreply@openai.com>

Author: Korayem

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

@@ -360,7 +360,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

@@ -23,6 +23,7 @@
 from ..schema import fq_table
 from .schema import (
     SCHEMA_VERSION,
+    BankRowsJSONEncoding,
     TransferCausalRelation,
     TransferChunk,
     TransferDocument,
@@ -272,7 +273,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
@@ -327,6 +334,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

@@ -30,6 +30,7 @@
 from ..schema import fq_table
 from .schema import (
     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,14 +314,10 @@ 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. A string may be serialized JSON from a raw export
-                # or an already-decoded JSON scalar from a codec-enabled export.
-                if isinstance(value, str):
-                    try:
-                        json.loads(value)
-                    except json.JSONDecodeError:
-                        value = json.dumps(value)
-                elif value is not None:
+                # 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")
@@ -360,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
 
@@ -382,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)
@@ -407,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), "
@@ -524,6 +559,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:
@@ -643,11 +687,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

@@ -23,6 +23,7 @@
 SCHEMA_VERSION = 1
 
 ObservationScopes = Literal["per_tag", "combined", "all_combinations", "shared"] | list[list[str]]
+BankRowsJSONEncoding = Literal["decoded", "serialized"]
 
 
 class TransferCausalRelation(BaseModel):
@@ -136,3 +137,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_document_transfer.py

@@ -128,12 +128,23 @@ def test_export_jsonb_coercion_preserves_decoded_scalar_string():
     assert _as_jsonb('{"scope": "combined"}') == {"scope": "combined"}
 
 
+def test_legacy_bank_archive_defaults_to_decoded_json_rows():
+    """Released v1 bank archives came from the codec-enabled admin CLI."""
+    from hindsight_api.engine.transfer.importer import _resolve_bank_rows_json_encoding
+
+    manifest = TransferManifest(source_bank_id="legacy", archive_type="bank")
+
+    assert manifest.bank_rows_json_encoding is None
+    assert _resolve_bank_rows_json_encoding(manifest) == "decoded"
+
+
 @pytest.mark.asyncio
 async def test_restore_rows_normalizes_jsonb_strings(memory):
-    """JSONB restore accepts decoded scalars without double-encoding JSON text."""
+    """JSONB restore follows archive provenance instead of guessing from strings."""
     from hindsight_api.engine.transfer.importer import _restore_rows
 
-    request_id = uuid.uuid4()
+    decoded_request_id = uuid.uuid4()
+    serialized_request_id = uuid.uuid4()
     backend = await memory._get_backend()
     async with acquire_with_retry(backend) as conn:
         try:
@@ -142,27 +153,48 @@ async def test_restore_rows_normalizes_jsonb_strings(memory):
                 "llm_requests",
                 [
                     {
-                        "id": str(request_id),
+                        "id": str(decoded_request_id),
                         "status": "completed",
                         "input": "I am an already-decoded scalar",
-                        "output": json.dumps("I am already-serialized JSON text"),
+                        "output": '{"answer":"JSON-looking decoded scalar"}',
                         "llm_info": {"shape": "decoded-object"},
-                        "metadata": json.dumps({"shape": "serialized-object"}),
                     }
                 ],
+                bank_rows_json_encoding="decoded",
             )
-            row = await conn.fetchrow(
-                f"SELECT input::text, output::text, llm_info::text, metadata::text "
-                f"FROM {fq_table('llm_requests')} WHERE id = $1",
-                request_id,
+            await _restore_rows(
+                conn,
+                "llm_requests",
+                [
+                    {
+                        "id": str(serialized_request_id),
+                        "status": "completed",
+                        "input": json.dumps("serialized scalar"),
+                        "output": json.dumps({"answer": "serialized object"}),
+                    }
+                ],
+                bank_rows_json_encoding="serialized",
+            )
+            decoded_row = await conn.fetchrow(
+                f"SELECT input::text, output::text, llm_info::text FROM {fq_table('llm_requests')} WHERE id = $1",
+                decoded_request_id,
             )
-            assert row is not None
-            assert json.loads(row["input"]) == "I am an already-decoded scalar"
-            assert json.loads(row["output"]) == "I am already-serialized JSON text"
-            assert json.loads(row["llm_info"]) == {"shape": "decoded-object"}
-            assert json.loads(row["metadata"]) == {"shape": "serialized-object"}
+            serialized_row = await conn.fetchrow(
+                f"SELECT input::text, output::text FROM {fq_table('llm_requests')} WHERE id = $1",
+                serialized_request_id,
+            )
+            assert decoded_row is not None
+            assert json.loads(decoded_row["input"]) == "I am an already-decoded scalar"
+            assert json.loads(decoded_row["output"]) == '{"answer":"JSON-looking decoded scalar"}'
+            assert json.loads(decoded_row["llm_info"]) == {"shape": "decoded-object"}
+            assert serialized_row is not None
+            assert json.loads(serialized_row["input"]) == "serialized scalar"
+            assert json.loads(serialized_row["output"]) == {"answer": "serialized object"}
         finally:
-            await conn.execute(f"DELETE FROM {fq_table('llm_requests')} WHERE id = $1", request_id)
+            await conn.execute(
+                f"DELETE FROM {fq_table('llm_requests')} WHERE id = ANY($1)",
+                [decoded_request_id, serialized_request_id],
+            )
 
 
 @pytest.mark.asyncio
@@ -197,6 +229,7 @@ async def test_export_bank_contents(memory, request_context):
             webhooks = json.loads(zf.read("webhooks.json"))
 
         assert manifest.archive_type == "bank"
+        assert manifest.bank_rows_json_encoding == "serialized"
         assert manifest.document_count == 1
         assert manifest.webhook_count == 1
         assert "mental_models.json" in names and "directives.json" in names
@@ -236,7 +269,7 @@ async def _bank_content_snapshot(memory, bank_id):
             f"SELECT name, disposition, mission, config FROM {fq_table('banks')} WHERE bank_id = $1", bank_id
         )
         docs = await conn.fetch(
-            f"SELECT id, original_text, tags FROM {fq_table('documents')} WHERE bank_id = $1", bank_id
+            f"SELECT id, original_text, tags, created_at FROM {fq_table('documents')} WHERE bank_id = $1", bank_id
         )
         facts = await conn.fetch(
             f"SELECT text, fact_type, context FROM {fq_table('memory_units')} "
@@ -268,7 +301,9 @@ async def _bank_content_snapshot(memory, bank_id):
         )
     return {
         "bank": (bank["name"], _as_json(bank["disposition"]), bank["mission"], _as_json(bank["config"])),
-        "documents": sorted((d["id"], d["original_text"], tuple(sorted(d["tags"] or []))) for d in docs),
+        "documents": sorted(
+            (d["id"], d["original_text"], tuple(sorted(d["tags"] or [])), d["created_at"]) for d in docs
+        ),
         "facts": sorted((f["text"], f["fact_type"], f["context"]) for f in facts),
         "observations": sorted((o["text"], o["proof_count"]) for o in obs),
         "entities": sorted(e["canonical_name"].lower() for e in ents),
@@ -678,9 +713,17 @@ async def test_export_import_observations(memory, request_context):
 
         # Create a real observation over those source facts.
         backend = await memory._get_backend()
+        archived_event_date = datetime(2001, 2, 3, 4, 5, 6, tzinfo=timezone.utc)
         async with acquire_with_retry(backend) as conn:
             async with conn.transaction():
                 await _create_observation_directly(conn, memory, src, source_ids, "Alice and Bob are colleagues.")
+                await conn.execute(
+                    f"UPDATE {fq_table('memory_units')} SET event_date = $1 "
+                    f"WHERE bank_id = $2 AND fact_type = 'observation' AND text = $3",
+                    archived_event_date,
+                    src,
+                    "Alice and Bob are colleagues.",
+                )
 
         # Export WITHOUT observations -> none in the archive (the bank may also
         # contain auto-consolidation observations; the flag is what gates them).
@@ -695,6 +738,7 @@ async def test_export_import_observations(memory, request_context):
         assert parsed.manifest.observation_count == len(parsed.observations) >= 1
         mine = next((o for o in parsed.observations if o.text == "Alice and Bob are colleagues."), None)
         assert mine is not None
+        assert mine.event_date == archived_event_date
         assert len(mine.sources) == 2  # both sources resolved within the export
         assert "embedding" not in archive.decode("utf-8", errors="ignore")
 
@@ -708,12 +752,13 @@ async def test_export_import_observations(memory, request_context):
         # and those source facts are marked consolidated.
         async with acquire_with_retry(backend) as conn:
             obs_row = await conn.fetchrow(
-                f"SELECT source_memory_ids FROM {fq_table('memory_units')} "
+                f"SELECT source_memory_ids, event_date FROM {fq_table('memory_units')} "
                 f"WHERE bank_id = $1 AND fact_type = 'observation' AND text = $2",
                 dst,
                 "Alice and Bob are colleagues.",
             )
             assert obs_row is not None
+            assert obs_row["event_date"] == archived_event_date
             dst_sources = list(obs_row["source_memory_ids"] or [])
             assert len(dst_sources) == 2
             consolidated = await conn.fetchval(