fix(sessions): import coerces archived-row flags and splits live rows once

A transfer row's active/compacted flags were read by truthiness, so a
hand-edited or foreign JSONL row with "active": "0" (truthy string) imported
live - putting compaction-archived turns back into model context - and
"active": null archived a live row. Coerce both flags with the same
_coerce_or(..., int, default) used for the int session columns: a missing,
null, or unparsable active flag means live (older exports carry none), and
compacted defaults to 0.

The same pass partitions rows into live and archived up front, replacing the
id()-set complement, the dead "_row_id" guard (_insert_message_rows always
sets it), and the subtract-and-reparse counter fixup: message_count and
tool_call_count now come straight from the live rows, with identical values
for well-formed exports. The kept round-trip test re-imports its payload with
stringified/null flags and asserts every row keeps its state.

Co-authored-by: John Paul Soliva <soliva.johnpaul@icloud.com>

hermes_state_portability.py

@@ -549,23 +549,25 @@ class SessionPortabilityMixin:
         sanitized_messages = [
             {**msg, **{key: _json_value(msg.get(key)) for key in _IMPORT_MESSAGE_JSON_FIELDS}} for msg in messages
         ]
-        total_messages, total_tool_calls = self._insert_message_rows(
-            conn, session_id, sanitized_messages, prune_checkpoints=False)
         # A row exported archived (``include_inactive``) must stay archived: inserted live, compacted or
-        # rewound turns would re-enter model context. Session counters count live rows only.
+        # rewound turns would re-enter model context. Flags are coerced like the int session columns, so a
-        archived = [msg for msg in sanitized_messages if "active" in msg and not msg["active"] and "_row_id" in msg]
+        # hand-edited "0" archives and a missing/null/unparsable flag imports live (older exports have none).
+        live: List[Dict[str, Any]] = []
+        archived: List[Dict[str, Any]] = []
+        for msg in sanitized_messages:
+            (live if self._coerce_or(msg.get("active"), int, 1) else archived).append(msg)
+        self._insert_message_rows(conn, session_id, sanitized_messages, prune_checkpoints=False)
         if archived:
             conn.executemany("UPDATE messages SET active = 0, compacted = ? WHERE id = ?",
-                             [(1 if msg.get("compacted") else 0, msg["_row_id"]) for msg in archived])
+                             [(1 if self._coerce_or(msg.get("compacted"), int, 0) else 0, msg["_row_id"])
-            total_messages -= len(archived)
+                              for msg in archived])
-            total_tool_calls -= sum(_tool_calls_count(_parse_tool_calls(msg.get("tool_calls"))) for msg in archived)
         # Pruning keys on live rows, so it runs only now: while every row was still live, an archived row's
         # newer checkpoint would strip the newest live one, and archived rows keep theirs as in the donor.
-        archived_ids = {id(msg) for msg in archived}
+        self._prune_shadowed_checkpoints(conn, session_id, live)
-        self._prune_shadowed_checkpoints(conn, session_id,
+        # Session counters count live rows only.
-                                         [msg for msg in sanitized_messages if id(msg) not in archived_ids])
         conn.execute("UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?",
-                     (total_messages, total_tool_calls, session_id))
+                     (len(live), sum(_tool_calls_count(_parse_tool_calls(msg.get("tool_calls"))) for msg in live),
+                      session_id))
 
     @staticmethod
     def _attach_import_parents(conn, parent_updates: List[tuple]) -> int:

tests/hermes_state/test_session_archived_round_trip.py

@@ -53,6 +53,18 @@ def test_export_all_round_trips_compacted_history(tmp_path):
         assert _shape(dst) == live_before, "archived turns must not re-enter live model context"
         # Session counters count live rows only.
         assert dst.get_session(STRANDED_ID)["message_count"] == len(live_before)
+
+        # Hand-edited/foreign JSONL: string flags ("0" is truthy) and null for live rows must not
+        # flip archived rows live or live rows archived.
+        for msg in payload[0]["messages"]:
+            msg["active"], msg["compacted"] = (str(msg["active"]) if not msg["active"] else None,
+                                               str(msg["compacted"]))
+        edited = SessionDB(db_path=tmp_path / "edited.db")
+        try:
+            assert edited.import_sessions(payload)["ok"]
+            assert _shape(edited, include_inactive=True) == all_before
+        finally:
+            edited.close()
     finally:
         src.close()
         dst.close()