diff --git a/memanto/cli/migrate/mappers.py b/memanto/cli/migrate/mappers.py index fd5f1d997..81a0181d1 100644 --- a/memanto/cli/migrate/mappers.py +++ b/memanto/cli/migrate/mappers.py @@ -55,15 +55,20 @@ } _DEFAULT_TITLE_CHARS = 80 +_MAX_TITLE_CHARS = 100 # MemoryRecord.title max_length _MAX_CONTENT_CHARS = 10000 # MemoryRecord.content max_length _MAX_FOOTER_CHARS = 800 # cap supporting-data footer so it never dominates +def _truncate_title(text: str, max_chars: int) -> str: + if len(text) <= max_chars: + return text + return text[: max_chars - 3].rstrip() + "..." + + def _title_from(content: str) -> str: text = content.strip().replace("\n", " ") - if len(text) <= _DEFAULT_TITLE_CHARS: - return text - return text[: _DEFAULT_TITLE_CHARS - 3].rstrip() + "..." + return _truncate_title(text, _DEFAULT_TITLE_CHARS) def _coerce_type(raw: str | None) -> str | None: @@ -455,6 +460,10 @@ def map_okf(export: dict[str, Any]) -> list[dict[str, Any]]: created_at = _parse_dt(entry.get("timestamp")) footer_items: list[tuple[str, Any]] = [ + ( + "Original OKF title", + title if len(title) > _MAX_TITLE_CHARS else None, + ), ("OKF source", entry.get("source_path")), # Only surface the OKF type when we couldn't map it to a slot. ("OKF type", okf_type if not memory_type else None), @@ -472,7 +481,11 @@ def map_okf(export: dict[str, Any]) -> list[dict[str, Any]]: rows.append( { - "title": title or _title_from(content), + "title": ( + _truncate_title(title, _MAX_TITLE_CHARS) + if title + else _title_from(content) + ), "content": content, "type": memory_type, "tags": tags, diff --git a/tests/test_okf.py b/tests/test_okf.py index e55bb5933..dc232496c 100644 --- a/tests/test_okf.py +++ b/tests/test_okf.py @@ -8,6 +8,7 @@ ``[Supporting data]`` footer without loss. """ +from memanto.app.core import MemoryRecord from memanto.app.services.okf_export_service import OkfExportService from memanto.cli.migrate.mappers import map_okf from memanto.cli.migrate.okf_loader import load_okf_bundle @@ -171,6 +172,25 @@ def test_foreign_okf_bundle_is_lossless(tmp_path): assert "customers -> /tables/customers.md" in row["content"] # link -> footer +def test_foreign_okf_long_title_stays_batch_valid(): + """An overlong foreign title must not invalidate its whole import batch.""" + long_title = "T" * 101 + export = { + "memories": [ + {"title": "Valid neighbor", "body": "This row should also import."}, + {"title": long_title, "body": "This title exceeds Memanto's limit."}, + ] + } + + rows = map_okf(export) + + assert rows[1]["title"] == ("T" * 97) + "..." + assert f"Original OKF title: {long_title}" in rows[1]["content"] + for row in rows: + payload = {key: value for key, value in row.items() if value is not None} + MemoryRecord(agent_id="agent1", actor_id="agent1", **payload) + + def test_loader_splits_stacked_file(tmp_path): """A stacked per-type file is split back into one entry per memory.""" memories_by_type = {