feat(gateway): resume interrupted sessions after restart

cli-config.yaml.example

@@ -395,6 +395,13 @@ session_reset:
 # explicitly want one shared "room brain" per group/channel.
 group_sessions_per_user: true
 
+# When true, a graceful gateway shutdown writes a small restart ledger for
+# in-flight sessions and the next gateway boot injects a hidden continuation
+# turn so interrupted work can resume from the last persisted context.
+# Sessions that were waiting on dangerous-command approval are resumed with a
+# note that approval state was lost and must be requested again if needed.
+resume_inflight_sessions_on_restart: false
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Gateway Streaming
 # ─────────────────────────────────────────────────────────────────────────────

gateway/config.py

@@ -253,6 +253,9 @@ class GatewayConfig:
     # Streaming configuration
     streaming: StreamingConfig = field(default_factory=StreamingConfig)
 
+    # Restart recovery
+    resume_inflight_sessions_on_restart: bool = False
+
     def get_connected_platforms(self) -> List[Platform]:
         """Return list of platforms that are enabled and configured."""
         connected = []
@@ -335,6 +338,7 @@ def to_dict(self) -> Dict[str, Any]:
             "group_sessions_per_user": self.group_sessions_per_user,
             "unauthorized_dm_behavior": self.unauthorized_dm_behavior,
             "streaming": self.streaming.to_dict(),
+            "resume_inflight_sessions_on_restart": self.resume_inflight_sessions_on_restart,
         }
 
     @classmethod
@@ -394,6 +398,9 @@ def from_dict(cls, data: Dict[str, Any]) -> "GatewayConfig":
             group_sessions_per_user=_coerce_bool(group_sessions_per_user, True),
             unauthorized_dm_behavior=unauthorized_dm_behavior,
             streaming=StreamingConfig.from_dict(data.get("streaming", {})),
+            resume_inflight_sessions_on_restart=_coerce_bool(
+                data.get("resume_inflight_sessions_on_restart"), False
+            ),
         )
 
     def get_unauthorized_dm_behavior(self, platform: Optional[Platform] = None) -> str:
@@ -483,6 +490,11 @@ def load_gateway_config() -> GatewayConfig:
                     "pair",
                 )
 
+            if "resume_inflight_sessions_on_restart" in yaml_cfg:
+                gw_data["resume_inflight_sessions_on_restart"] = yaml_cfg[
+                    "resume_inflight_sessions_on_restart"
+                ]
+
             # Merge platforms section from config.yaml into gw_data so that
             # nested keys like platforms.webhook.extra.routes are loaded.
             yaml_platforms = yaml_cfg.get("platforms")

gateway/platforms/base.py

@@ -343,6 +343,10 @@ class MessageEvent:
 
     # Auto-loaded skill for topic/channel bindings (e.g., Telegram DM Topics)
     auto_skill: Optional[str] = None
+
+    # Optional clean text to persist in transcripts when ``text`` contains
+    # internal routing hints or restart-resume prefixes.
+    persist_user_message: Optional[str] = None
 
     # Timestamps
     timestamp: datetime = field(default_factory=datetime.now)