diff --git a/hermes_state.py b/hermes_state.py
index bdbd93c3debba..9bdbcdbd8e4c9 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -50,7 +50,7 @@
     _watched_sqlite_sidecar_paths, is_zeroed_state_db, quarantine_cross_process_lock, quarantine_zeroed_state_db,
     refuse_deleted_wal_generation,
 )
-from hermes_state_messages import SessionMessagesMixin
+from hermes_state_messages import SessionMessagesMixin, _ensure_session_row
 from hermes_state_wal import _WAL_INCOMPAT_MARKERS, apply_database_pragmas, apply_wal_with_fallback
 from hermes_state_repair import _claim_repair_attempt, preflight_db_writability, repair_state_db_schema
 from hermes_state_titles import SessionTitlesMixin
diff --git a/hermes_state_messages.py b/hermes_state_messages.py
index 07cb4c69ee57e..5ab173e0f564f 100644
--- a/hermes_state_messages.py
+++ b/hermes_state_messages.py
@@ -42,6 +42,27 @@
 _INVALID = object()  # _json_or sentinel where the fallback must be distinguishable from JSON null
 
 
+def _ensure_session_row(conn, session_id: str) -> None:
+    """Ensure the session parent row exists to satisfy foreign key constraints.
+
+    A missing row (deleted by cleanup or never created after a transient
+    create_session failure) would otherwise raise FOREIGN KEY constraint
+    failed and abort the whole turn or batch flush. Stamped with
+    source='self-healed' so maintenance tooling can identify and filter it.
+    """
+    cur = conn.execute(
+        "INSERT OR IGNORE INTO sessions (id, source, started_at) "
+        "VALUES (?, 'self-healed', ?)",
+        (session_id, time.time()),
+    )
+    if cur.rowcount > 0:
+        logger.warning(
+            "Self-healed missing parent session row for session_id=%s (source='self-healed')",
+            session_id,
+        )
+
+
+
 def _json_or(raw: Any, fallback: Any, warning: str) -> Any:
     """``json.loads(raw)``; on failure log *warning* and return *fallback*."""
     try:
@@ -283,6 +304,8 @@ def append_message(
         def _do(conn):
             self._check_transcript_write_guards(conn, session_id, compression_lock_holder,
                 turn_lease_holder=turn_lease_holder, turn_lease_ttl_seconds=turn_lease_ttl_seconds)
+            # FK self-heal: ensure the session parent row exists.
+            _ensure_session_row(conn, session_id)
             msg_id = conn.execute(_INSERT_MESSAGE_SQL, params).lastrowid
             self._bump_session_counters(conn, session_id, 1, _tool_calls_count(tool_calls), unit=True)
             return msg_id
@@ -306,6 +329,8 @@ def append_messages_batch(
         def _do(conn):
             self._check_transcript_write_guards(conn, session_id, compression_lock_holder,
                 turn_lease_holder=turn_lease_holder, turn_lease_ttl_seconds=turn_lease_ttl_seconds)
+            # FK self-heal: same guarantee as append_message for batch flush.
+            _ensure_session_row(conn, session_id)
             from agent.transcript_repair import resolve_and_repair_transcript_batch
             inserted_rows = resolve_and_repair_transcript_batch(conn, session_id, messages,
                 encode_content_fn=self._encode_content, decode_content_fn=self._decode_content)
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index bb676562b9aca..b0f7edf0f2b9a 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -5852,3 +5852,38 @@ def test_percent_preserved_for_cjk_query(self):
         # text; keep % intact there (pre-existing contract).
         sanitized = self._sanitize("完成50%")
         assert "%" in sanitized
+
+
+class TestSessionFKSelfHeal:
+    """Verify that transcript appends self-heal missing parent session rows."""
+
+    def test_append_message_self_heals_missing_session(self, tmp_path):
+        import hermes_state
+        db = hermes_state.SessionDB(db_path=tmp_path / "state.db")
+        # append_message to a session that was never created
+        msg_id = db.append_message("orphan_s1", role="user", content="hello orphan")
+        assert msg_id is not None
+        session = db.get_session("orphan_s1")
+        assert session is not None
+        assert session["id"] == "orphan_s1"
+        assert session["source"] == "self-healed"
+        assert session["message_count"] == 1
+
+    def test_append_messages_batch_self_heals_missing_session(self, tmp_path):
+        import hermes_state
+        db = hermes_state.SessionDB(db_path=tmp_path / "state.db")
+        # append_messages_batch to an uncreated session
+        inserted = db.append_messages_batch(
+            "orphan_s2",
+            [
+                {"role": "user", "content": "batch 1"},
+                {"role": "assistant", "content": "batch 2"},
+            ],
+        )
+        assert inserted == 2
+        session = db.get_session("orphan_s2")
+        assert session is not None
+        assert session["id"] == "orphan_s2"
+        assert session["source"] == "self-healed"
+        assert session["message_count"] == 2
+