Ruff

demo/demo.py

@@ -85,7 +85,9 @@ async def main() -> None:
                 data.total = 99
                 raise ValueError("simulated failure")
         except ValueError:
-            print(f"\nReset rolled back: {data.total=} (we can always read data without tx)\n")
+            print(
+                f"\nReset rolled back: {data.total=} (we can always read data without tx)\n"
+            )
 
         with kanta.transaction(action="delete", user="userid002") as data:
             del data.users["userid001"]

kanta/logging.py

@@ -187,6 +187,11 @@ def _join_path(path: str, key: str) -> str:
     return f"{path}.{key}"
 
 
+def _dim_ellipsis() -> str:
+    """Return the truncation ellipsis in the palette's ellipsis color."""
+    return str(Line().ellipsis("…"))
+
+
 def _format_value(
     value: Any,
     path: str,
@@ -209,7 +214,7 @@ def _format_value(
     if isinstance(value, str):
         value = _UNSAFE_CHARS.sub("", value)
         if len(value) > max_len:
-            return value[: max_len - 3] + "..."
+            return value[: max_len - 1] + _dim_ellipsis()
         return value
     if isinstance(value, dict):
         if not value:
@@ -235,7 +240,7 @@ def _format_value(
         return "[" + ", ".join(parts) + "]"
     text = str(value)
     if len(text) > max_len:
-        text = text[: max_len - 3] + "..."
+        text = text[: max_len - 1] + _dim_ellipsis()
     return text
 
 
@@ -361,15 +366,21 @@ def _format_change_lines(
         path_str = _format_path(path, logfmt, final_color="add")
         if isinstance(value, dict) and value:
             lines = [str(Line()("  ", path_str, " ").sep("="))]
-            formatted_items = []
             base_path = ".".join(path)
-            for k, v in value.items():
+            keys = []
+            for k in value:
                 key_path = _join_path(base_path, str(k))
-                key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
+                keys.append((k, _format_value(k, key_path, max_len=30, logfmt=logfmt)))
-                v_str = _format_value(v, key_path, max_len=30, logfmt=logfmt)
+            field_width = max(displaywidth(kd) for _, kd in keys)
-                formatted_items.append((key_display, v_str))
-            field_width = max(displaywidth(k) for k, _ in formatted_items)
             field_width = max(field_width, 12)
+            # Each item line is "    {key:{field_width}}: {value}"; budget the
+            # value so the whole line fits in 80 columns.
+            value_width = max(80 - 4 - field_width - 2, 20)
+            formatted_items = []
+            for (k, key_display), v in zip(keys, value.values()):
+                key_path = _join_path(base_path, str(k))
+                v_str = _format_value(v, key_path, max_len=value_width, logfmt=logfmt)
+                formatted_items.append((key_display, v_str))
             return lines + [
                 str(
                     Line()("    ", k).sep(":")(

kanta/tty.py

@@ -75,6 +75,7 @@ class Colors:
     path_final = "38;5;250"  # White for the final path element
     add = "32"  # Green for additions
     delete = "1;31"  # Bold red for deletions
+    ellipsis = "38;5;242"  # Dark grey for the truncation ellipsis
 
 
 colors = Colors()

tests/test_kanta_integration.py

@@ -562,13 +562,12 @@ async def test_migration_with_changes_records_diff_and_snapshot(
 
     records = read_changes(path, format_config)
     migration_records = [r for r in records if r.a.startswith("migrate")]
-    assert len(migration_records) == 2
+    # The version migration and the msgspec normalization that follows it are
+    # grouped into a single migrate:vN record.
+    assert len(migration_records) == 1
     assert migration_records[0].a == "migrate:v1"
     assert migration_records[0].v == 1
-    assert migration_records[0].diff == {"counter": 2}
+    assert migration_records[0].diff == {"counter": 2, "users": {}}
-    assert migration_records[1].a == "migrate:msgspec"
-    assert migration_records[1].v == 1
-    assert migration_records[1].diff == {"users": {}}
 
     snap = read_last_snapshot(path, format_config)
     assert snap is not None