feat(session): lazy session creation — defer SQLite write until first message

32.4% of all sessions (3,564 of 10,985) were empty — created but never
used. The root cause: get_or_create_session() wrote to SQLite
immediately, even for webhook pings and platform handshake events that
never produced a real conversation.

Changes:
- Added _db_persisted flag to SessionEntry (transient, not serialized)
- get_or_create_session() now marks new sessions as _db_persisted=False
  instead of calling db.create_session()
- New ensure_db_session() method flushes the DB record lazily on first
  real user message
- Gateway _handle_message_with_agent() calls ensure_db_session() before
  dispatching to the agent
- reset_session() also defers DB create for consistency

Impact: Eliminates ~3,500 wasted session records per cycle. Sessions
that never receive a message never hit the DB.

Refs #314

gateway/run.py

@@ -2304,6 +2304,9 @@ class GatewayRunner:
         # Get or create session
         session_entry = self.session_store.get_or_create_session(source)
         session_key = session_entry.session_key
+
+        # Lazy session creation: persist to SQLite on first real message
+        self.session_store.ensure_db_session(session_entry)
         
         # Emit session:start for new or auto-reset sessions
         _is_new_session = (

gateway/session.py

@@ -383,6 +383,11 @@ class SessionEntry:
     # survives gateway restarts (the old in-memory _pre_flushed_sessions
     # set was lost on restart, causing redundant re-flushes).
     memory_flushed: bool = False
+
+    # Lazy session creation: tracks whether the session record has been
+    # written to SQLite.  New sessions start False; the DB write is
+    # deferred until the first user message arrives.
+    _db_persisted: bool = True
     
     def to_dict(self) -> Dict[str, Any]:
         result = {
@@ -763,11 +768,10 @@ class SessionStore:
             except Exception as e:
                 logger.debug("Session DB operation failed: %s", e)
 
-        if self._db and db_create_kwargs:
-            try:
-                self._db.create_session(**db_create_kwargs)
-            except Exception as e:
-                print(f"[gateway] Warning: Failed to create SQLite session: {e}")
+        # Lazy session creation: defer DB write until first user message.
+        # Mark the entry as not yet persisted; ensure_db_session() will
+        # flush it when the gateway receives an actual message.
+        entry._db_persisted = False
 
         # Seed new DM thread sessions with parent DM session history.
         # When a bot reply creates a Slack thread and the user responds in it,
@@ -806,6 +810,26 @@ class SessionStore:
 
         return entry
 
+    def ensure_db_session(self, entry: SessionEntry) -> None:
+        """Lazily persist a session to SQLite on first user message.
+
+        Called by the gateway message handler when a real message arrives.
+        If the session is already persisted, this is a no-op.
+        """
+        if entry._db_persisted or not self._db:
+            return
+        try:
+            source_val = entry.platform.value if entry.platform else "unknown"
+            user_id = entry.origin.user_id if entry.origin else None
+            self._db.create_session(
+                session_id=entry.session_id,
+                source=source_val,
+                user_id=user_id,
+            )
+            entry._db_persisted = True
+        except Exception as e:
+            logger.warning("Failed to lazily create SQLite session: %s", e)
+
     def update_session(
         self,
         session_key: str,
@@ -865,11 +889,8 @@ class SessionStore:
             except Exception as e:
                 logger.debug("Session DB operation failed: %s", e)
 
-        if self._db and db_create_kwargs:
-            try:
-                self._db.create_session(**db_create_kwargs)
-            except Exception as e:
-                logger.debug("Session DB operation failed: %s", e)
+        # Lazy: defer DB create until first message
+        new_entry._db_persisted = False
 
         return new_entry