hermes: harden chat continuity and activity

services/hermes/chat-statefulset.yaml

@@ -204,6 +204,9 @@ spec:
             - |
               /opt/hermes/.venv/bin/python /opt/coordinator/patch_api_server_sessions.py \
                 /opt/hermes/gateway/platforms/api_server.py /patched/api_server.py
+              grep -Fq 'conversation_history = compact_telegram_history(conversation_history)' /patched/api_server.py
+              grep -Fq 'full_history = compact_telegram_history(full_history)' /patched/api_server.py
+              grep -Fq 'conversation_history_snapshot = compact_telegram_history(' /patched/api_server.py
               /opt/hermes/.venv/bin/python /opt/coordinator/migrate_telegram_api_sessions.py \
                 /opt/data/state.db /opt/data/response_store.db
           securityContext:
@@ -284,6 +287,7 @@ spec:
             - {name: runtime-access, mountPath: /runtime-access}
             - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py}
             - {name: api-server-patch, mountPath: /opt/hermes/gateway/platforms/api_server.py, subPath: api_server.py}
+            - {name: coordinator, mountPath: /opt/hermes/gateway/platforms/telegram_continuity.py, subPath: migrate_telegram_api_sessions.py, readOnly: true}
             - {name: stream-recovery-patch, mountPath: /opt/hermes/agent/conversation_loop.py, subPath: conversation_loop.py}
             - {name: subprocess-secret-patch, mountPath: /opt/hermes/tools/environments/local.py, subPath: local.py}
             - {name: subprocess-secret-patch, mountPath: /opt/hermes/tools/process_registry.py, subPath: process_registry.py}

services/hermes/router/main.go

@@ -27,8 +27,9 @@ type linkRecord struct {
 }
 
 type telegramTopic struct {
-	Label    string `json:"label"`
+	Label      string `json:"label"`
-	LastUsed int64  `json:"last_used"`
+	LastUsed   int64  `json:"last_used"`
+	Generation int64  `json:"generation,omitempty"`
 }
 
 type telegramTopicState struct {
@@ -372,7 +373,12 @@ func (router *tenantRouter) ServeHTTP(writer http.ResponseWriter, request *http.
 		return
 	}
 	proxy := httputil.NewSingleHostReverseProxy(target)
-	proxy.ModifyResponse = injectChatBridge
+	proxy.ModifyResponse = func(response *http.Response) error {
+		if err := boundSessionSnapshot(response); err != nil {
+			return err
+		}
+		return injectChatBridge(response)
+	}
 	proxy.ErrorHandler = func(writer http.ResponseWriter, _ *http.Request, proxyErr error) {
 		log.Printf("tenant slot %d unavailable", slot)
 		http.Error(writer, "private chat runtime is starting", http.StatusBadGateway)

services/hermes/router/main_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"encoding/json"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -126,6 +127,9 @@ func TestRouterProxiesWebUIAndAddsTelegramShortcut(t *testing.T) {
 	if !strings.Contains(response.Body.String(), "hermes-chat-bridge.js") || !strings.Contains(response.Body.String(), "hermes-chat-bridge.css") {
 		t.Fatal("Telegram shortcut assets were not injected")
 	}
+	if !strings.Contains(response.Body.String(), "hermes-session-continuity.js") {
+		t.Fatal("session continuity fallback was not injected")
+	}
 	assetRequest := httptest.NewRequest(http.MethodGet, "/hermes-chat-bridge.js", nil)
 	assetRequest.Header.Set("X-Forwarded-User", "subject")
 	assetResponse := httptest.NewRecorder()
@@ -145,6 +149,70 @@ func TestRouterProxiesWebUIAndAddsTelegramShortcut(t *testing.T) {
 	}
 }
 
+func TestSessionContinuityAssetHasAccessiblePollFallback(t *testing.T) {
+	router, err := newTenantRouter(filepath.Join(t.TempDir(), "state.json"), 1, func(int) string { return "" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := httptest.NewRequest(http.MethodGet, "/hermes-session-continuity.js", nil)
+	request.Header.Set("X-Forwarded-User", "subject")
+	response := httptest.NewRecorder()
+	router.ServeHTTP(response, request)
+	if response.Code != http.StatusOK {
+		t.Fatalf("got status %d", response.Code)
+	}
+	asset := response.Body.String()
+	for _, expected := range []string{
+		"aria-live", "aria-busy", "/api/sessions/", "/messages?limit=24&hermes_fallback=1",
+		"Latest stored activity", "no renderable messages",
+		"Session updates disconnected", "oauth2/start?rd=", "payload.session_id",
+		"session.ended_at == null", "latest.observed",
+	} {
+		if !strings.Contains(asset, expected) {
+			t.Fatalf("session fallback omitted %q", expected)
+		}
+	}
+	if strings.Contains(asset, "location.replace") || strings.Contains(asset, "/api/session?") {
+		t.Fatal("session fallback discards the active session")
+	}
+}
+
+func TestSessionContinuityPollIsBoundedByRouter(t *testing.T) {
+	backend := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		messages := make([]map[string]int, 30)
+		for index := range messages {
+			messages[index] = map[string]int{"index": index}
+		}
+		writer.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(writer).Encode(map[string]any{
+			"session_id": "resolved", "messages": messages,
+		})
+	}))
+	defer backend.Close()
+	router, err := newTenantRouter(filepath.Join(t.TempDir(), "state.json"), 1, func(int) string { return backend.URL })
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := httptest.NewRequest(
+		http.MethodGet,
+		"/api/sessions/root/messages?limit=24&hermes_fallback=1",
+		nil,
+	)
+	request.Header.Set("X-Forwarded-User", "subject")
+	response := httptest.NewRecorder()
+	router.ServeHTTP(response, request)
+	if response.Code != http.StatusOK {
+		t.Fatalf("got status %d", response.Code)
+	}
+	var payload sessionSnapshot
+	if err := json.NewDecoder(response.Body).Decode(&payload); err != nil {
+		t.Fatal(err)
+	}
+	if len(payload.Messages) != sessionSnapshotItems || payload.TotalMessages != 30 {
+		t.Fatalf("router returned unbounded fallback: %#v", payload)
+	}
+}
+
 func TestTelegramPageExplainsOneTimeOperatorSetup(t *testing.T) {
 	router, err := newTenantRouter(filepath.Join(t.TempDir(), "state.json"), 1, func(slot int) string { return "" })
 	if err != nil {

services/hermes/router/session_continuity.go

@@ -0,0 +1,97 @@
+package main
+
+const sessionContinuityJS = `(() => {
+  const match = location.pathname.match(/^\/session\/([^/]+)\/?$/);
+  if (!match) return;
+  let sessionId;
+  try { sessionId = decodeURIComponent(match[1]); } catch (_) { sessionId = match[1]; }
+  const started = Date.now();
+  let timer = 0;
+  let request = null;
+  const card = document.createElement('aside');
+  card.id = 'hermes-session-continuity';
+  card.setAttribute('role', 'status');
+  card.setAttribute('aria-live', 'polite');
+  card.setAttribute('aria-atomic', 'true');
+  card.hidden = true;
+  card.style.cssText = 'position:fixed;right:16px;bottom:16px;z-index:1000;max-width:min(420px,calc(100vw - 32px));box-sizing:border-box;padding:10px 12px;border:1px solid #475569;border-radius:10px;background:#111827;color:#e5e7eb;font:13px/1.4 system-ui,sans-serif;box-shadow:0 8px 28px #0008';
+  const message = document.createElement('span');
+  const newChat = document.createElement('a');
+  newChat.href = '/';
+  newChat.textContent = ' Start a new chat.';
+  newChat.style.color = '#7dd3fc';
+  card.append(message, newChat);
+  document.body.appendChild(card);
+
+  const show = (text, busy, link) => {
+    message.textContent = text;
+    card.hidden = false;
+    card.setAttribute('aria-busy', busy ? 'true' : 'false');
+    newChat.hidden = !link;
+  };
+  const hide = () => { card.hidden = true; card.setAttribute('aria-busy', 'false'); };
+  const activityLabel = (messages) => {
+    const latest = messages[messages.length - 1] || {};
+    if (latest.tool_name) return 'tool ' + String(latest.tool_name).slice(0, 80);
+    if (Array.isArray(latest.tool_calls) && latest.tool_calls.length) {
+      const call = latest.tool_calls[latest.tool_calls.length - 1] || {};
+      return 'tool ' + String((call.function || {}).name || 'activity').slice(0, 80);
+    }
+    if (latest.activity_event) return String(latest.activity_event).replace(/[._-]+/g, ' ').slice(0, 80);
+    if (latest.observed && typeof latest.content === 'string') return latest.content.slice(0, 120);
+    return latest.role === 'assistant' ? 'assistant update' : latest.role === 'user' ? 'request stored' : 'working';
+  };
+  const authOrMissing = (response) => {
+    if (response.status === 401 || response.status === 403) {
+      const rd = location.pathname + location.search + location.hash;
+      location.assign('/oauth2/start?rd=' + encodeURIComponent(rd));
+      return 'auth';
+    }
+    if (response.status === 404) {
+      show('This session is unavailable to this account.', false, true);
+      schedule(10000);
+      return 'missing';
+    }
+    return '';
+  };
+  const schedule = (delay) => {
+    clearTimeout(timer);
+    timer = window.setTimeout(poll, delay);
+  };
+  async function poll() {
+    if (request) return;
+    request = new AbortController();
+    const timeout = window.setTimeout(() => request && request.abort(), 8000);
+    try {
+      const root = '/api/sessions/' + encodeURIComponent(sessionId);
+      const messageResponse = await fetch(root + '/messages?limit=24&hermes_fallback=1', {cache:'no-store', credentials:'same-origin', signal:request.signal});
+      if (authOrMissing(messageResponse)) return;
+      if (!messageResponse.ok) throw new Error('session messages poll failed');
+      const payload = await messageResponse.json();
+      const messages = Array.isArray(payload.messages) ? payload.messages : [];
+      const resolvedId = typeof payload.session_id === 'string' && payload.session_id ? payload.session_id : sessionId;
+      const detailResponse = await fetch('/api/sessions/' + encodeURIComponent(resolvedId), {cache:'no-store', credentials:'same-origin', signal:request.signal});
+      if (authOrMissing(detailResponse)) return;
+      if (!detailResponse.ok) throw new Error('session detail poll failed');
+      const session = await detailResponse.json();
+      const lastActive = Number(session.last_active || session.started_at || 0);
+      const durableWorker = session.source === 'api_server' && Boolean(session.parent_session_id);
+      const recentlyInteractive = Number.isFinite(lastActive) && Date.now() / 1000 - lastActive < 300;
+      const active = session.ended_at == null && messages.length > 0 && (durableWorker || recentlyInteractive);
+      if (active) show('Hermes is working. Latest stored activity: ' + activityLabel(messages) + '.', true, false);
+      else if (!messages.length && Date.now() - started >= 4000) show('This session has no renderable messages yet. It may be new or no longer available.', false, true);
+      else hide();
+      schedule(document.hidden ? 10000 : 2500);
+    } catch (_) {
+      show('Session updates disconnected. Retrying without changing this session…', true, false);
+      schedule(document.hidden ? 10000 : 3000);
+    } finally {
+      clearTimeout(timeout);
+      request = null;
+    }
+  }
+  addEventListener('online', () => schedule(0));
+  addEventListener('pageshow', () => schedule(0));
+  document.addEventListener('visibilitychange', () => schedule(0));
+  schedule(1200);
+})();`

services/hermes/router/session_snapshot.go

@@ -0,0 +1,65 @@
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+)
+
+const (
+	sessionSnapshotItems = 24
+	sessionSnapshotBytes = 8 << 20
+)
+
+type sessionSnapshot struct {
+	SessionID     string            `json:"session_id"`
+	Messages      []json.RawMessage `json:"messages"`
+	TotalMessages int               `json:"total_messages"`
+}
+
+// boundSessionSnapshot caps only the continuity fallback response. Native
+// WebUI requests remain untouched, including when an older backend ignores
+// its optional `limit` query parameter.
+func boundSessionSnapshot(response *http.Response) error {
+	request := response.Request
+	if request == nil || response.StatusCode != http.StatusOK ||
+		request.URL.Query().Get("hermes_fallback") != "1" ||
+		!strings.HasPrefix(request.URL.Path, "/api/sessions/") ||
+		!strings.HasSuffix(request.URL.Path, "/messages") {
+		return nil
+	}
+	body, err := io.ReadAll(io.LimitReader(response.Body, sessionSnapshotBytes+1))
+	if err != nil {
+		return err
+	}
+	_ = response.Body.Close()
+	if len(body) > sessionSnapshotBytes {
+		return errors.New("session snapshot exceeds safe response limit")
+	}
+	var payload sessionSnapshot
+	if err := json.Unmarshal(body, &payload); err != nil || payload.Messages == nil {
+		return errors.New("session snapshot is malformed")
+	}
+	total := len(payload.Messages)
+	if payload.TotalMessages > total {
+		total = payload.TotalMessages
+	}
+	if len(payload.Messages) > sessionSnapshotItems {
+		payload.Messages = payload.Messages[len(payload.Messages)-sessionSnapshotItems:]
+	}
+	payload.TotalMessages = total
+	body, err = json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	response.Body = io.NopCloser(strings.NewReader(string(body)))
+	response.ContentLength = int64(len(body))
+	response.Header.Set("Content-Length", strconv.Itoa(len(body)))
+	response.Header.Set("Content-Type", "application/json; charset=utf-8")
+	response.Header.Set("Cache-Control", "no-store")
+	response.Header.Del("ETag")
+	return nil
+}

services/hermes/router/session_snapshot_test.go

@@ -0,0 +1,113 @@
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"testing/iotest"
+)
+
+func snapshotResponse(target, body string) *http.Response {
+	request, _ := http.NewRequest(http.MethodGet, target, nil)
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     make(http.Header),
+		Body:       io.NopCloser(strings.NewReader(body)),
+		Request:    request,
+	}
+}
+
+func TestBoundSessionSnapshotKeepsOnlyRecentMessages(t *testing.T) {
+	messages := make([]map[string]int, 30)
+	for index := range messages {
+		messages[index] = map[string]int{"index": index}
+	}
+	body, _ := json.Marshal(map[string]any{
+		"session_id": "resolved", "messages": messages,
+	})
+	response := snapshotResponse(
+		"http://tenant/api/sessions/root/messages?limit=24&hermes_fallback=1",
+		string(body),
+	)
+	response.Header.Set("ETag", "stale")
+	if err := boundSessionSnapshot(response); err != nil {
+		t.Fatal(err)
+	}
+	var bounded sessionSnapshot
+	if err := json.NewDecoder(response.Body).Decode(&bounded); err != nil {
+		t.Fatal(err)
+	}
+	if len(bounded.Messages) != 24 || bounded.TotalMessages != 30 {
+		t.Fatalf("snapshot was not bounded: %#v", bounded)
+	}
+	var first map[string]int
+	if err := json.Unmarshal(bounded.Messages[0], &first); err != nil || first["index"] != 6 {
+		t.Fatalf("snapshot did not retain the recent tail: %#v, %v", first, err)
+	}
+	if response.Header.Get("Cache-Control") != "no-store" || response.Header.Get("ETag") != "" {
+		t.Fatal("bounded snapshot retained cache metadata")
+	}
+}
+
+func TestBoundSessionSnapshotPreservesLargerReportedTotal(t *testing.T) {
+	response := snapshotResponse(
+		"http://tenant/api/sessions/root/messages?hermes_fallback=1",
+		`{"session_id":"leaf","messages":[],"total_messages":100}`,
+	)
+	if err := boundSessionSnapshot(response); err != nil {
+		t.Fatal(err)
+	}
+	var bounded sessionSnapshot
+	if err := json.NewDecoder(response.Body).Decode(&bounded); err != nil {
+		t.Fatal(err)
+	}
+	if bounded.SessionID != "leaf" || bounded.TotalMessages != 100 {
+		t.Fatalf("reported total was lost: %#v", bounded)
+	}
+}
+
+func TestBoundSessionSnapshotFailsClosedOnMalformedOrOversizedBody(t *testing.T) {
+	for name, body := range map[string]string{
+		"malformed": `{`,
+		"oversized": strings.Repeat("x", sessionSnapshotBytes+1),
+	} {
+		t.Run(name, func(t *testing.T) {
+			response := snapshotResponse(
+				"http://tenant/api/sessions/root/messages?hermes_fallback=1",
+				body,
+			)
+			if err := boundSessionSnapshot(response); err == nil {
+				t.Fatal("unsafe snapshot was accepted")
+			}
+		})
+	}
+	response := snapshotResponse(
+		"http://tenant/api/sessions/root/messages?hermes_fallback=1", `{}`,
+	)
+	response.Body = io.NopCloser(iotest.ErrReader(errors.New("read failed")))
+	if err := boundSessionSnapshot(response); err == nil {
+		t.Fatal("snapshot body read failure was ignored")
+	}
+}
+
+func TestBoundSessionSnapshotLeavesNativeAndErrorResponsesUntouched(t *testing.T) {
+	for _, response := range []*http.Response{
+		{StatusCode: http.StatusOK},
+		snapshotResponse("http://tenant/api/sessions/root/messages", `{}`),
+		snapshotResponse("http://tenant/api/sessions/root", `{}`),
+	} {
+		if err := boundSessionSnapshot(response); err != nil {
+			t.Fatal(err)
+		}
+	}
+	errorResponse := snapshotResponse(
+		"http://tenant/api/sessions/root/messages?hermes_fallback=1", `{}`,
+	)
+	errorResponse.StatusCode = http.StatusNotFound
+	if err := boundSessionSnapshot(errorResponse); err != nil {
+		t.Fatal(err)
+	}
+}

services/hermes/router/telegram.go

@@ -24,20 +24,21 @@ type telegramConfig struct {
 }
 
 type telegramBot struct {
-	config       telegramConfig
+	config        telegramConfig
-	router       *tenantRouter
+	router        *tenantRouter
-	apiBase      string
+	apiBase       string
-	fileBase     string
+	fileBase      string
-	client       *http.Client
+	client        *http.Client
-	mediaClient  *http.Client
+	mediaClient   *http.Client
-	agentClient  *http.Client
+	agentClient   *http.Client
-	mu           sync.RWMutex
+	mu            sync.RWMutex
-	botUsername  string
+	botUsername   string
-	ready        bool
+	ready         bool
-	lastError    string
+	lastError     string
-	workLimit    chan struct{}
+	workLimit     chan struct{}
-	slotLocks    map[int]*sync.Mutex
+	activityEvery time.Duration
-	slotLocksMux sync.Mutex
+	slotLocks     map[int]*sync.Mutex
+	slotLocksMux  sync.Mutex
 }
 
 type telegramUpdate struct {
@@ -96,15 +97,16 @@ func readTelegramConfig(path string) (telegramConfig, error) {
 
 func newTelegramBot(config telegramConfig, router *tenantRouter) *telegramBot {
 	return &telegramBot{
-		config:      config,
+		config:        config,
-		router:      router,
+		router:        router,
-		apiBase:     "https://api.telegram.org/bot" + config.BotToken,
+		apiBase:       "https://api.telegram.org/bot" + config.BotToken,
-		fileBase:    "https://api.telegram.org/file/bot" + config.BotToken,
+		fileBase:      "https://api.telegram.org/file/bot" + config.BotToken,
-		client:      &http.Client{Timeout: 70 * time.Second},
+		client:        &http.Client{Timeout: 70 * time.Second},
-		mediaClient: newTenantMediaClient(),
+		mediaClient:   newTenantMediaClient(),
-		agentClient: &http.Client{Timeout: 15 * time.Minute},
+		agentClient:   &http.Client{Timeout: 15 * time.Minute},
-		workLimit:   make(chan struct{}, 4),
+		workLimit:     make(chan struct{}, 4),
-		slotLocks:   map[int]*sync.Mutex{},
+		activityEvery: 4 * time.Second,
+		slotLocks:     map[int]*sync.Mutex{},
 	}
 }
 
@@ -238,16 +240,6 @@ func (bot *telegramBot) run() {
 	}
 }
 
-func commandParts(text string) (string, []string) {
-	fields := strings.Fields(strings.TrimSpace(text))
-	if len(fields) == 0 || !strings.HasPrefix(fields[0], "/") {
-		return "", nil
-	}
-	command := strings.TrimPrefix(strings.ToLower(fields[0]), "/")
-	command, _, _ = strings.Cut(command, "@")
-	return command, fields[1:]
-}
-
 func (bot *telegramBot) handleUpdate(update telegramUpdate) {
 	message := update.Message
 	if message == nil || message.From == nil || message.Chat.Type != "private" || message.Chat.ID != message.From.ID {
@@ -276,7 +268,7 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) {
 		return
 	}
 	if command == "help" {
-		_ = bot.sendText(message.Chat.ID, "Send text or a photo to chat with Hermes. Use /topic <name> to create or switch a durable topic, /topics to list topics, or /unlink to disconnect this Telegram account. Model and intensity controls are available in Hermes WebUI.")
+		_ = bot.sendText(message.Chat.ID, "Send text or a photo to chat with Hermes. Photos are available for analysis on the message that carries them. Image edits can reuse the latest image Hermes generated; inbound Telegram photos are not stored as editable artifacts. Use /topic <name> to create or switch a durable topic, /topics to list topics, or /unlink to disconnect this Telegram account. Model and intensity controls are available in Hermes WebUI.")
 		return
 	}
 	slot, linked := bot.router.telegramSlot(userID)
@@ -332,7 +324,9 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) {
 		_ = bot.sendText(message.Chat.ID, "Hermes could not save the active topic right now. Please try again shortly.")
 		return
 	}
-	_ = bot.sendAction(message.Chat.ID, "typing")
+	text = telegramInputPrompt(text, hasPhoto)
+	stopActivity := bot.startTelegramActivity(message.Chat.ID)
+	defer stopActivity()
 	var reply string
 	if hasPhoto {
 		media, fetchErr := bot.fetchTelegramPhoto(message.Photo)
@@ -429,6 +423,10 @@ func (bot *telegramBot) askTenantRequest(slot int, bodyReader io.Reader, updateI
 func (bot *telegramBot) sendAction(chatID int64, action string) error {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
+	return bot.sendActionContext(ctx, chatID, action)
+}
+
+func (bot *telegramBot) sendActionContext(ctx context.Context, chatID int64, action string) error {
 	return bot.call(ctx, "sendChatAction", url.Values{
 		"chat_id": {strconv.FormatInt(chatID, 10)},
 		"action":  {action},

services/hermes/router/telegram_input.go

@@ -12,6 +12,7 @@ import (
 	"os"
 	"path"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -22,6 +23,93 @@ type telegramInputMedia struct {
 	Path string
 }
 
+func commandParts(text string) (string, []string) {
+	fields := strings.Fields(strings.TrimSpace(text))
+	if len(fields) == 0 || !strings.HasPrefix(fields[0], "/") {
+		return "", nil
+	}
+	command := strings.TrimPrefix(strings.ToLower(fields[0]), "/")
+	command, _, _ = strings.Cut(command, "@")
+	return command, fields[1:]
+}
+
+const inboundTelegramEditNotice = "Transport note: this Telegram photo is available for visual analysis on this turn, but it is not stored as a generated-image artifact. The image_edit_latest tools can reuse only an image Hermes generated. Do not claim an exact edit of this inbound photo; explain the limitation and offer either a new image inspired by it or an edit of the latest Hermes-generated image."
+
+func looksLikeImageEdit(text string) bool {
+	lower := strings.ToLower(strings.Join(strings.Fields(text), " "))
+	for _, phrase := range []string{
+		"edit this", "edit the image", "edit the photo", "change this image",
+		"change this photo", "modify this", "retouch", "turn this into",
+		"turn this cat", "turn this dog", "remove from the image",
+		"replace in the image", "edit it", "change it", "modify it", "crop it",
+		"resize it", "upscale it",
+	} {
+		if strings.Contains(lower, phrase) {
+			return true
+		}
+	}
+	hasTarget := false
+	for _, target := range []string{"image", "photo", "picture", "portrait", "selfie", "screenshot"} {
+		if strings.Contains(lower, target) {
+			hasTarget = true
+			break
+		}
+	}
+	if hasTarget {
+		for _, action := range []string{
+			"add ", "blur", "brighten", "change", "color", "convert", "crop",
+			"edit", "make ", "modify", "remove", "replace", "resize", "sharpen",
+			"transform", "turn ", "upscale",
+		} {
+			if strings.Contains(lower, action) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func telegramInputPrompt(text string, hasPhoto bool) string {
+	if hasPhoto {
+		return inboundTelegramEditNotice + "\n\nUser request: " + text
+	}
+	if !looksLikeImageEdit(text) {
+		return text
+	}
+	return "Transport note: no image is attached to this Telegram message. Use image_edit_latest only if Hermes previously generated an image in this private tenant. Otherwise do not claim access to an earlier inbound Telegram photo; offer a new image or ask the user to use Hermes WebUI with the source attachment.\n\nUser request: " + text
+}
+
+func (bot *telegramBot) startTelegramActivity(chatID int64) func() {
+	done := make(chan struct{})
+	stopped := make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
+	every := bot.activityEvery
+	if every <= 0 {
+		every = 4 * time.Second
+	}
+	go func() {
+		defer close(stopped)
+		ticker := time.NewTicker(every)
+		defer ticker.Stop()
+		for {
+			_ = bot.sendActionContext(ctx, chatID, "typing")
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			close(done)
+			cancel()
+		})
+		<-stopped
+	}
+}
+
 func largestTelegramPhoto(photos []telegramPhotoSize) (telegramPhotoSize, error) {
 	var selected telegramPhotoSize
 	for _, photo := range photos {

services/hermes/router/telegram_media.go

@@ -24,7 +24,11 @@ const (
 	tenantMediaLimit     = 50 << 20
 )
 
-var telegramMediaMarker = regexp.MustCompile(`MEDIA:([^\s\)\]]+)`)
+var (
+	telegramMediaMarker = regexp.MustCompile(`(?i)MEDIA:\s*([^\s\)\]\}"']+)`)
+	telegramMediaLabel  = regexp.MustCompile(`(?i)MEDIA:`)
+	telegramPrivatePath = regexp.MustCompile(`(?i)(?:/opt/data/(?:cache/images|workspace)|/workspace)/[^\s\)\]\}"']+`)
+)
 
 type telegramReply struct {
 	Text       string
@@ -50,6 +54,8 @@ func parseTelegramReply(text string) telegramReply {
 		}
 	}
 	cleaned := telegramMediaMarker.ReplaceAllString(text, "")
+	cleaned = telegramPrivatePath.ReplaceAllString(cleaned, "[private attachment]")
+	cleaned = telegramMediaLabel.ReplaceAllString(cleaned, "")
 	lines := strings.Split(cleaned, "\n")
 	compact := make([]string, 0, len(lines))
 	blank := false

services/hermes/router/telegram_media_test.go

@@ -124,6 +124,20 @@ func TestParseTelegramReplyRemovesOneOrMoreInternalMarkers(t *testing.T) {
 	}
 }
 
+func TestParseTelegramReplyScrubsMalformedMarkersAndBarePrivatePaths(t *testing.T) {
+	reply := parseTelegramReply(
+		"Delivered. MEDIA: /opt/data/cache/images/one.png and /workspace/private/two.png; MEDIA:",
+	)
+	if len(reply.MediaPaths) != 1 || reply.MediaPaths[0] != "/opt/data/cache/images/one.png" {
+		t.Fatalf("valid spaced marker was not retained for delivery: %#v", reply)
+	}
+	for _, unsafe := range []string{"MEDIA:", "/opt/data/", "/workspace/"} {
+		if strings.Contains(reply.Text, unsafe) {
+			t.Fatalf("unsafe transport detail %q leaked in %q", unsafe, reply.Text)
+		}
+	}
+}
+
 func TestNormalizeTenantMediaPath(t *testing.T) {
 	allowed := map[string]string{
 		"/workspace/./renders//portrait.png":          "/opt/data/workspace/renders/portrait.png",

services/hermes/router/telegram_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -273,6 +274,144 @@ func TestTelegramTopicValidationAndLegacyGeneralConversation(t *testing.T) {
 	}
 }
 
+func TestTelegramTopicContinuityExpiresOnlyAfterIdleWeek(t *testing.T) {
+	statePath := filepath.Join(t.TempDir(), "state.json")
+	base := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)
+	now := base
+	router, err := newTenantRouter(statePath, 1, func(int) string { return "" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	router.now = func() time.Time { return now }
+	first, err := router.selectTelegramTopic("123", "Week plan")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, elapsed := range []time.Duration{time.Hour, 24 * time.Hour} {
+		now = base.Add(elapsed)
+		resumed, touchErr := router.touchTelegramTopic("123")
+		if touchErr != nil || resumed.Conversation != first.Conversation {
+			t.Fatalf("topic changed after %s: %#v, %v", elapsed, resumed, touchErr)
+		}
+	}
+	// Exactly seven idle days remains the same durable conversation.
+	now = base.Add(8 * 24 * time.Hour)
+	exactWeek, err := router.touchTelegramTopic("123")
+	if err != nil || exactWeek.Conversation != first.Conversation {
+		t.Fatalf("topic changed at exact idle week: %#v, %v", exactWeek, err)
+	}
+	reloaded, err := newTenantRouter(statePath, 1, func(int) string { return "" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	reloaded.now = func() time.Time { return now }
+	if active := reloaded.activeTelegramTopic("123"); active != exactWeek {
+		t.Fatalf("router restart changed topic: %#v != %#v", active, exactWeek)
+	}
+	now = now.Add(7*24*time.Hour + time.Second)
+	rotated, err := reloaded.touchTelegramTopic("123")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rotated.Conversation == first.Conversation || !strings.HasSuffix(rotated.Conversation, "-g1") {
+		t.Fatalf("expired topic did not rotate deterministically: %#v", rotated)
+	}
+	again, err := newTenantRouter(statePath, 1, func(int) string { return "" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	if active := again.activeTelegramTopic("123"); active != rotated {
+		t.Fatalf("rotated identity changed after restart: %#v != %#v", active, rotated)
+	}
+}
+
+func TestTelegramActivityRepeatsAndStops(t *testing.T) {
+	var actions atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if request.URL.Path != "/sendChatAction" {
+			t.Fatalf("unexpected Telegram method %s", request.URL.Path)
+		}
+		actions.Add(1)
+		writer.Header().Set("Content-Type", "application/json")
+		_, _ = writer.Write([]byte(`{"ok":true,"result":true}`))
+	}))
+	defer server.Close()
+	bot := &telegramBot{
+		apiBase:       server.URL,
+		client:        &http.Client{Timeout: time.Second},
+		activityEvery: 5 * time.Millisecond,
+	}
+	stop := bot.startTelegramActivity(42)
+	deadline := time.Now().Add(time.Second)
+	for actions.Load() < 2 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	stop()
+	stoppedAt := actions.Load()
+	time.Sleep(20 * time.Millisecond)
+	if stoppedAt < 2 || actions.Load() != stoppedAt {
+		t.Fatalf("activity did not progress and stop: before=%d after=%d", stoppedAt, actions.Load())
+	}
+}
+
+func TestTelegramActivityStopCancelsInflightRequest(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, request *http.Request) {
+		close(started)
+		select {
+		case <-request.Context().Done():
+		case <-release:
+		}
+	}))
+	defer server.Close()
+	bot := &telegramBot{
+		apiBase:       server.URL,
+		client:        &http.Client{Timeout: time.Minute},
+		activityEvery: time.Hour,
+	}
+	stop := bot.startTelegramActivity(42)
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("activity request did not start")
+	}
+	stopped := make(chan struct{})
+	go func() {
+		stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		t.Fatal("activity stop did not cancel the in-flight request")
+	}
+	close(release)
+}
+
+func TestTelegramImageEditPromptStatesTransportBoundary(t *testing.T) {
+	withPhoto := telegramInputPrompt("Edit this photo into a watercolor", true)
+	if !strings.Contains(withPhoto, "available for visual analysis") ||
+		!strings.Contains(withPhoto, "not stored as a generated-image artifact") {
+		t.Fatalf("inbound attachment boundary missing: %q", withPhoto)
+	}
+	withoutPhoto := telegramInputPrompt("Edit this photo into a watercolor", false)
+	if !strings.Contains(withoutPhoto, "no image is attached") ||
+		!strings.Contains(withoutPhoto, "image_edit_latest only if Hermes previously generated") {
+		t.Fatalf("prior generated image boundary missing: %q", withoutPhoto)
+	}
+	for _, request := range []string{"Make the photo brighter", "Crop it closer"} {
+		if got := telegramInputPrompt(request, false); !strings.Contains(got, "no image is attached") {
+			t.Fatalf("edit variant omitted attachment boundary: %q", got)
+		}
+	}
+	plain := telegramInputPrompt("Explain this photograph", true)
+	if !strings.Contains(plain, "available for visual analysis") ||
+		!strings.Contains(plain, "User request: Explain this photograph") {
+		t.Fatalf("photo analysis boundary missing: %q", plain)
+	}
+}
+
 func TestTelegramInputRejectsUnsafePathsAndNonImages(t *testing.T) {
 	for _, value := range []string{"", "../secret", "/absolute/image.jpg", `photos\image.jpg`, "photos/../../secret"} {
 		if _, err := safeTelegramFilePath(value); err == nil {

services/hermes/router/telegram_topics.go

@@ -5,12 +5,16 @@ import (
 	"encoding/hex"
 	"errors"
 	"sort"
+	"strconv"
 	"strings"
 	"unicode"
 	"unicode/utf8"
 )
 
-const telegramDefaultTopic = "General"
+const (
+	telegramDefaultTopic = "General"
+	telegramTopicIdleTTL = 7 * 24 * 60 * 60
+)
 
 type activeTelegramTopic struct {
 	Conversation string
@@ -42,12 +46,35 @@ func telegramTopicID(label string) string {
 }
 
 func telegramTopicConversation(topicID string) string {
+	return telegramTopicConversationGeneration(topicID, 0)
+}
+
+func telegramTopicConversationGeneration(topicID string, generation int64) string {
 	if topicID == "" || topicID == "general" {
 		// Preserve the original named conversation so existing Telegram context
 		// survives rollout into topic-aware routing.
-		return "telegram"
+		if generation == 0 {
+			return "telegram"
+		}
+		topicID = "general"
 	}
-	return "telegram-topic-" + topicID
+	conversation := "telegram-topic-" + topicID
+	if generation > 0 {
+		conversation += "-g" + strconv.FormatInt(generation, 10)
+	}
+	return conversation
+}
+
+func topicExpired(topic telegramTopic, now int64) bool {
+	return topic.LastUsed > 0 && now > topic.LastUsed && now-topic.LastUsed > telegramTopicIdleTTL
+}
+
+func touchTopic(topic telegramTopic, now int64) telegramTopic {
+	if topicExpired(topic, now) {
+		topic.Generation++
+	}
+	topic.LastUsed = now
+	return topic
 }
 
 func (router *tenantRouter) activeTelegramTopic(userID string) activeTelegramTopic {
@@ -63,7 +90,10 @@ func (router *tenantRouter) activeTelegramTopic(userID string) activeTelegramTop
 	if topic, found := state.Topics[topicID]; found && strings.TrimSpace(topic.Label) != "" {
 		label = topic.Label
 	}
-	return activeTelegramTopic{Conversation: telegramTopicConversation(topicID), Label: label}
+	return activeTelegramTopic{
+		Conversation: telegramTopicConversationGeneration(topicID, state.Topics[topicID].Generation),
+		Label:        label,
+	}
 }
 
 func (router *tenantRouter) selectTelegramTopic(userID, rawLabel string) (activeTelegramTopic, error) {
@@ -83,12 +113,18 @@ func (router *tenantRouter) selectTelegramTopic(userID, rawLabel string) (active
 		state.Topics = map[string]telegramTopic{}
 	}
 	state.Active = topicID
-	state.Topics[topicID] = telegramTopic{Label: label, LastUsed: router.now().Unix()}
+	topic := state.Topics[topicID]
+	topic.Label = label
+	topic = touchTopic(topic, router.now().Unix())
+	state.Topics[topicID] = topic
 	router.state.TelegramTopics[identity] = state
 	if err := router.saveLocked(); err != nil {
 		return activeTelegramTopic{}, err
 	}
-	return activeTelegramTopic{Conversation: telegramTopicConversation(topicID), Label: label}, nil
+	return activeTelegramTopic{
+		Conversation: telegramTopicConversationGeneration(topicID, topic.Generation),
+		Label:        label,
+	}, nil
 }
 
 func (router *tenantRouter) touchTelegramTopic(userID string) (activeTelegramTopic, error) {
@@ -108,13 +144,16 @@ func (router *tenantRouter) touchTelegramTopic(userID string) (activeTelegramTop
 	if strings.TrimSpace(topic.Label) == "" {
 		topic.Label = telegramDefaultTopic
 	}
-	topic.LastUsed = router.now().Unix()
+	topic = touchTopic(topic, router.now().Unix())
 	state.Topics[topicID] = topic
 	router.state.TelegramTopics[identity] = state
 	if err := router.saveLocked(); err != nil {
 		return activeTelegramTopic{}, err
 	}
-	return activeTelegramTopic{Conversation: telegramTopicConversation(topicID), Label: topic.Label}, nil
+	return activeTelegramTopic{
+		Conversation: telegramTopicConversationGeneration(topicID, topic.Generation),
+		Label:        topic.Label,
+	}, nil
 }
 
 func (router *tenantRouter) telegramTopics(userID string) []telegramTopic {

services/hermes/router/web.go

@@ -354,6 +354,15 @@ func (router *tenantRouter) serveTelegramWeb(writer http.ResponseWriter, request
 		writer.Header().Set("Cache-Control", "public, max-age=3600")
 		_, _ = io.WriteString(writer, bridgeJS)
 		return true
+	case "/hermes-session-continuity.js":
+		if request.Method != http.MethodGet {
+			http.Error(writer, "method not allowed", http.StatusMethodNotAllowed)
+			return true
+		}
+		writer.Header().Set("Content-Type", "application/javascript; charset=utf-8")
+		writer.Header().Set("Cache-Control", "public, max-age=3600")
+		_, _ = io.WriteString(writer, sessionContinuityJS)
+		return true
 	case "/telegram":
 		if request.Method != http.MethodGet {
 			http.Error(writer, "method not allowed", http.StatusMethodNotAllowed)
@@ -455,6 +464,9 @@ func injectChatBridge(response *http.Response) error {
 		content = strings.Replace(content, "</head>", `<link rel="stylesheet" href="/hermes-chat-bridge.css?v=20260813-telegram-readiness-v1"></head>`, 1)
 		content = strings.Replace(content, "</body>", `<script src="/hermes-chat-bridge.js?v=20260813-telegram-readiness-v1" defer></script></body>`, 1)
 	}
+	if !strings.Contains(content, "hermes-session-continuity.js") {
+		content = strings.Replace(content, "</body>", `<script src="/hermes-session-continuity.js?v=20260817-v1" defer></script></body>`, 1)
+	}
 	response.Body = io.NopCloser(strings.NewReader(content))
 	response.ContentLength = int64(len(content))
 	response.Header.Set("Content-Length", strconv.Itoa(len(content)))

services/hermes/scripts/migrate_telegram_api_sessions.py

@@ -1,16 +1,236 @@
 #!/usr/bin/env python3
-"""Backfill Telegram origin metadata for durable API conversations."""
+"""Bound Telegram context and backfill durable API-session metadata."""
 
 from __future__ import annotations
 
 import argparse
 import json
+import re
 import sqlite3
 from pathlib import Path
+from typing import Any
+
+
+MAX_RESPONSE_BYTES = 32 << 20
+MAX_HISTORY_ITEMS = 80
+MAX_HISTORY_BYTES = 192_000
+MAX_RECENT_TURNS = 12
+MAX_ITEM_CHARS = 8_000
+MAX_SUMMARY_CHARS = 12_000
+SUMMARY_PREFIX = "Telegram continuity summary (older bounded context):"
+ATTACHMENT_NOTE = (
+    "[Earlier Telegram image omitted from bounded context; attach it again "
+    "in Hermes WebUI if the original pixels are required.]"
+)
+_CONVERSATION = re.compile(r"^telegram(?:-topic-[a-z0-9-]{1,160})?$")
+_MEDIA_PATH = re.compile(
+    r'(?:MEDIA:\s*)?(?:/opt/data/(?:cache/images|workspace)|/workspace)/[^\s\]\[)}`"]+',
+    re.IGNORECASE,
+)
+_DATA_IMAGE = re.compile(r"data:image/[^;,\s]+;base64,[A-Za-z0-9+/=]+", re.IGNORECASE)
+_MESSAGE_IDENTITY_COLUMNS = (
+    "session_id",
+    "role",
+    "content",
+    "tool_call_id",
+    "tool_calls",
+    "tool_name",
+    "timestamp",
+    "token_count",
+    "finish_reason",
+    "reasoning",
+    "reasoning_content",
+    "reasoning_details",
+    "codex_reasoning_items",
+    "codex_message_items",
+    "platform_message_id",
+    "observed",
+    "active",
+    "compacted",
+)
+
+
+def _text(value: Any) -> str:
+    """Return bounded context text without reusable transport-only images."""
+    if isinstance(value, str):
+        text = value
+    elif isinstance(value, list):
+        parts: list[str] = []
+        for part in value:
+            if not isinstance(part, dict):
+                continue
+            kind = str(part.get("type") or "").lower()
+            if "image" in kind or "image_url" in part:
+                parts.append(ATTACHMENT_NOTE)
+                continue
+            candidate = part.get("text", part.get("content", ""))
+            if isinstance(candidate, str):
+                parts.append(candidate)
+        text = "\n".join(parts)
+    else:
+        try:
+            text = json.dumps(value, ensure_ascii=False, sort_keys=True)
+        except (TypeError, ValueError):
+            text = str(value)
+    text = _DATA_IMAGE.sub(ATTACHMENT_NOTE, text)
+    text = _MEDIA_PATH.sub("[private attachment]", text)
+    return text[:MAX_ITEM_CHARS]
+
+
+def _normalized_item(raw: Any) -> dict[str, Any] | None:
+    if not isinstance(raw, dict):
+        return None
+    role = str(raw.get("role") or "").strip().lower()
+    if role not in {"assistant", "system", "tool", "user"}:
+        return None
+    content = _text(raw.get("content", ""))
+    if not content and role != "assistant":
+        return None
+    item: dict[str, Any] = {"role": role, "content": content}
+    if raw.get("_db_persisted") is True:
+        item["_db_persisted"] = True
+    for key in ("name", "tool_call_id", "tool_name"):
+        value = raw.get(key)
+        if isinstance(value, str) and value:
+            item[key] = value[:256]
+    calls = raw.get("tool_calls")
+    if isinstance(calls, list):
+        bounded_calls = []
+        for call in calls[:16]:
+            if not isinstance(call, dict):
+                continue
+            function = (
+                call.get("function") if isinstance(call.get("function"), dict) else {}
+            )
+            bounded = {
+                "id": str(call.get("id") or "")[:256],
+                "type": str(call.get("type") or "function")[:32],
+                "function": {
+                    "name": str(function.get("name") or "")[:256],
+                    "arguments": _text(function.get("arguments", ""))[:2_000],
+                },
+            }
+            bounded_calls.append(bounded)
+        if bounded_calls:
+            item["tool_calls"] = bounded_calls
+    return item
+
+
+def _fingerprint(item: dict[str, Any]) -> str:
+    try:
+        visible = {key: value for key, value in item.items() if not key.startswith("_")}
+        return json.dumps(
+            visible, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        )
+    except (TypeError, ValueError):
+        return repr(item)
+
+
+def _deduplicate(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Drop replay duplicates while preserving intentional repeated prose."""
+    result: list[dict[str, Any]] = []
+    stable_ids: set[tuple[str, str]] = set()
+    for item in items:
+        stable = ""
+        if isinstance(item.get("tool_call_id"), str):
+            stable = item["tool_call_id"]
+        elif isinstance(item.get("id"), str):
+            stable = item["id"]
+        if stable:
+            key = (item["role"], stable)
+            if key in stable_ids:
+                continue
+            stable_ids.add(key)
+        if result and _fingerprint(result[-1]) == _fingerprint(item):
+            continue
+        result.append(item)
+    # Response replays can append one exact transcript to itself. Remove only
+    # complete adjacent blocks, not ordinary repeated user phrases.
+    changed = True
+    while changed and len(result) > 1:
+        changed = False
+        for width in range(len(result) // 2, 0, -1):
+            if [_fingerprint(x) for x in result[-2 * width : -width]] == [
+                _fingerprint(x) for x in result[-width:]
+            ]:
+                del result[-width:]
+                changed = True
+                break
+    return result
+
+
+def _summary(older: list[dict[str, Any]]) -> dict[str, str] | None:
+    lines: list[str] = []
+    for item in older:
+        role = item["role"]
+        content = str(item.get("content") or "").strip()
+        if role == "system" and content.startswith(SUMMARY_PREFIX):
+            candidates = content[len(SUMMARY_PREFIX) :].strip().splitlines()
+        elif role in {"user", "assistant"} and content:
+            candidates = [f"- {role.title()}: {content[:600]}"]
+        else:
+            continue
+        for line in candidates:
+            clean = " ".join(line.split())
+            if clean and clean not in lines:
+                lines.append(clean)
+    content = SUMMARY_PREFIX
+    for line in lines[-40:]:
+        candidate = content + "\n" + line
+        if len(candidate) > MAX_SUMMARY_CHARS:
+            break
+        content = candidate
+    return (
+        {
+            "role": "system",
+            "content": content,
+            "_db_persisted": True,
+            "_compressed_summary": True,
+        }
+        if lines
+        else None
+    )
+
+
+def compact_telegram_history(history: Any) -> list[dict[str, Any]]:
+    """Return a deterministic summary plus bounded recent Telegram turns."""
+    if not isinstance(history, list):
+        return []
+    normalized = _deduplicate(
+        [item for raw in history if (item := _normalized_item(raw)) is not None]
+    )
+    recent: list[dict[str, Any]] = []
+    recent_bytes = 2
+    user_turns = 0
+    split = len(normalized)
+    for index in range(len(normalized) - 1, -1, -1):
+        item = normalized[index]
+        encoded = len(_fingerprint(item).encode("utf-8")) + 1
+        next_turns = user_turns + (item["role"] == "user")
+        if recent and (
+            len(recent) >= MAX_HISTORY_ITEMS
+            or recent_bytes + encoded > MAX_HISTORY_BYTES
+            or next_turns > MAX_RECENT_TURNS
+        ):
+            break
+        recent.insert(0, item)
+        recent_bytes += encoded
+        user_turns = next_turns
+        split = index
+    summary = _summary(normalized[:split])
+    result = ([summary] if summary else []) + recent
+    while (
+        len(json.dumps(result, ensure_ascii=False).encode("utf-8"))
+        > (MAX_HISTORY_BYTES + MAX_SUMMARY_CHARS)
+        and len(recent) > 1
+    ):
+        recent.pop(0)
+        result = ([summary] if summary else []) + recent
+    return result
 
 
 def telegram_sessions(response_store: Path) -> dict[str, str]:
-    """Return session ID to stable Telegram conversation-key mappings."""
+    """Return safe session ID to stable Telegram conversation mappings."""
     if not response_store.exists():
         return {}
     connection = sqlite3.connect(f"file:{response_store}?mode=ro", uri=True)
@@ -19,21 +239,64 @@ def telegram_sessions(response_store: Path) -> dict[str, str]:
             """SELECT c.name, r.data
                  FROM conversations AS c
                  JOIN responses AS r ON r.response_id = c.response_id
-                WHERE c.name = 'telegram' OR c.name LIKE 'telegram-topic-%'"""
+                WHERE (c.name = 'telegram' OR c.name LIKE 'telegram-topic-%')
+                  AND LENGTH(r.data) <= ?
+                ORDER BY c.name, r.response_id""",
+            (MAX_RESPONSE_BYTES,),
         ).fetchall()
+    except sqlite3.DatabaseError:
+        return {}
     finally:
         connection.close()
     result: dict[str, str] = {}
     for conversation, raw in rows:
-        try:
+        name = str(conversation or "")
-            session_id = str(json.loads(raw).get("session_id") or "").strip()
+        if not _CONVERSATION.fullmatch(name):
-        except (TypeError, ValueError, json.JSONDecodeError):
             continue
-        if session_id:
+        try:
-            result[session_id] = str(conversation)
+            payload = json.loads(raw)
+            session_id = str(payload.get("session_id") or "").strip()
+        except (AttributeError, TypeError, ValueError, json.JSONDecodeError):
+            continue
+        if (
+            session_id
+            and len(session_id) <= 256
+            and not re.search(r"[\r\n\x00/\\]", session_id)
+        ):
+            result.setdefault(session_id, name)
     return result
 
 
+def deduplicate_stored_messages(
+    connection: sqlite3.Connection, session_ids: list[str]
+) -> int:
+    """Remove only byte-for-byte replay rows for known Telegram API sessions."""
+    columns = {str(row[1]) for row in connection.execute("PRAGMA table_info(messages)")}
+    required = {"id", "session_id", "role", "content", "timestamp"}
+    if not required.issubset(columns):
+        return 0
+    identity = [column for column in _MESSAGE_IDENTITY_COLUMNS if column in columns]
+    partition = ", ".join(identity)
+    changed = 0
+    for session_id in session_ids:
+        query = f"""DELETE FROM messages WHERE id IN (
+                    SELECT id FROM (
+                        SELECT id, ROW_NUMBER() OVER (
+                            PARTITION BY {partition} ORDER BY id
+                        ) AS replay_number
+                        FROM messages
+                        WHERE session_id = ? AND EXISTS (
+                            SELECT 1 FROM sessions
+                            WHERE sessions.id = messages.session_id
+                              AND sessions.source = 'api_server'
+                        )
+                    ) WHERE replay_number > 1
+                )"""  # noqa: S608 -- identifiers come only from the fixed allowlist
+        cursor = connection.execute(query, (session_id,))
+        changed += max(cursor.rowcount, 0)
+    return changed
+
+
 def migrate(state_database: Path, response_store: Path) -> int:
     """Annotate known sessions without creating or rewriting conversations."""
     mappings = telegram_sessions(response_store)
@@ -43,12 +306,16 @@ def migrate(state_database: Path, response_store: Path) -> int:
     changed = 0
     try:
         columns = {
-            str(row[1])
+            str(row[1]) for row in connection.execute("PRAGMA table_info(sessions)")
-            for row in connection.execute("PRAGMA table_info(sessions)").fetchall()
         }
         required = {
-            "id", "source", "session_key", "chat_type", "display_name",
+            "id",
-            "origin_json", "title",
+            "source",
+            "session_key",
+            "chat_type",
+            "display_name",
+            "origin_json",
+            "title",
         }
         if not required.issubset(columns):
             return 0
@@ -57,23 +324,38 @@ def migrate(state_database: Path, response_store: Path) -> int:
                 {"platform": "telegram", "session_key": conversation},
                 separators=(",", ":"),
             )
-            title = "Telegram · General" if conversation == "telegram" else "Telegram"
+            default_title = (
+                "Telegram · General" if conversation == "telegram" else "Telegram"
+            )
             cursor = connection.execute(
                 """UPDATE sessions
-                      SET session_key = COALESCE(session_key, ?),
+                      SET session_key = ?, chat_type = 'private',
-                          chat_type = COALESCE(chat_type, 'private'),
+                          display_name = 'Telegram', origin_json = ?,
-                          display_name = COALESCE(display_name, 'Telegram'),
+                          title = CASE
-                          origin_json = COALESCE(origin_json, ?),
+                            WHEN title IS NULL OR TRIM(title) = '' OR
-                          title = COALESCE(title, ?)
+                                 LOWER(TRIM(title)) = 'unassigned'
+                            THEN ? ELSE title END
                     WHERE id = ? AND source = 'api_server'
-                      AND (
+                      AND (COALESCE(session_key, '') != ? OR
-                          session_key IS NULL OR chat_type IS NULL OR
+                           COALESCE(chat_type, '') != 'private' OR
-                          display_name IS NULL OR origin_json IS NULL OR title IS NULL
+                           COALESCE(display_name, '') != 'Telegram' OR
-                      )""",
+                           COALESCE(origin_json, '') != ? OR title IS NULL OR
-                (conversation, origin, title, session_id),
+                           TRIM(title) = '' OR LOWER(TRIM(title)) = 'unassigned')""",
+                (
+                    conversation,
+                    origin,
+                    default_title,
+                    session_id,
+                    conversation,
+                    origin,
+                ),
             )
             changed += max(cursor.rowcount, 0)
+        deduplicate_stored_messages(connection, list(mappings))
         connection.commit()
+    except sqlite3.DatabaseError:
+        connection.rollback()
+        return 0
     finally:
         connection.close()
     return changed

services/hermes/scripts/patch_api_server_sessions.py

@@ -6,15 +6,13 @@ from __future__ import annotations
 import argparse
 from pathlib import Path
 
-
+BEFORE = """        model = body.get("model") or self._model_name
-BEFORE = '''        model = body.get("model") or self._model_name
         system_prompt = body.get("system_prompt")
         if system_prompt is not None and not isinstance(system_prompt, str):
             return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
         db.create_session(session_id, "api_server", model=str(model) if model else None, system_prompt=system_prompt)
-'''
+"""
-
+AFTER = """        model = body.get("model") or self._model_name
-AFTER = '''        model = body.get("model") or self._model_name
         system_prompt = body.get("system_prompt")
         if system_prompt is not None and not isinstance(system_prompt, str):
             return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
@@ -46,21 +44,18 @@ AFTER = '''        model = body.get("model") or self._model_name
             system_prompt=system_prompt,
             parent_session_id=parent_session_id,
         )
-'''
+"""
-
+RUNS_BEFORE = """        run_id = f"run_{uuid.uuid4().hex}"
-RUNS_BEFORE = '''        run_id = f"run_{uuid.uuid4().hex}"
         session_id = body.get("session_id") or stored_session_id or run_id
         # Approval queues gate host-side tool execution and must be isolated
-'''
+"""
-
+RESPONSES_SESSION_BEFORE = """        # Reuse session from previous_response_id chain so the dashboard
-RESPONSES_SESSION_BEFORE = '''        # Reuse session from previous_response_id chain so the dashboard
         # groups the entire conversation under one session entry.
         session_id = stored_session_id or str(uuid.uuid4())
 
         # Per-client model routing for /v1/responses (see model_routes).
-'''
+"""
-
+RESPONSES_SESSION_AFTER = """        # Reuse session from previous_response_id chain so the dashboard
-RESPONSES_SESSION_AFTER = '''        # Reuse session from previous_response_id chain so the dashboard
         # groups the entire conversation under one session entry.
         session_id = stored_session_id or str(uuid.uuid4())
 
@@ -68,9 +63,7 @@ RESPONSES_SESSION_AFTER = '''        # Reuse session from previous_response_id c
         # WebUI do not collapse it into an anonymous Api_Server session. Keep
         # the accepted vocabulary narrow: these headers are presentation and
         # routing metadata, never an authorization boundary.
-        conversation_platform = request.headers.get(
+        # X-Hermes-Conversation-Platform was parsed before history compaction.
-            "X-Hermes-Conversation-Platform", ""
-        ).strip().lower()
         conversation_title = request.headers.get(
             "X-Hermes-Conversation-Title", ""
         ).strip()
@@ -117,9 +110,41 @@ RESPONSES_SESSION_AFTER = '''        # Reuse session from previous_response_id c
                 db.set_session_title(session_id, conversation_title)
 
         # Per-client model routing for /v1/responses (see model_routes).
-'''
+"""
+TRUNCATION_BEFORE = """        # Truncation support
+        if body.get("truncation") == "auto" and len(conversation_history) > 100:
+            conversation_history = conversation_history[-100:]
+"""
+TRUNCATION_AFTER = """        # Telegram keeps a durable compact summary plus bounded recent turns.
+        conversation_platform = request.headers.get(
+            "X-Hermes-Conversation-Platform", ""
+        ).strip().lower()
+        if conversation_platform == "telegram":
+            from gateway.platforms.telegram_continuity import compact_telegram_history
+            conversation_history = compact_telegram_history(conversation_history)
+        elif body.get("truncation") == "auto" and len(conversation_history) > 100:
+            conversation_history = conversation_history[-100:]
+"""
+HISTORY_STORE_BEFORE = """        # Build output items from the current turn only.  AIAgent returns a
+"""
+HISTORY_STORE_AFTER = """        if conversation_platform == "telegram":
+            full_history = compact_telegram_history(full_history)
 
-RUNS_AFTER = '''        run_id = f"run_{uuid.uuid4().hex}"
+        # Build output items from the current turn only.  AIAgent returns a
+"""
+
+SSE_STORE_BEFORE = """            self._response_store.put(response_id, {
+"""
+
+SSE_STORE_AFTER = """            if gateway_session_key and gateway_session_key.startswith("telegram"):
+                from gateway.platforms.telegram_continuity import compact_telegram_history
+                conversation_history_snapshot = compact_telegram_history(
+                    conversation_history_snapshot
+                )
+            self._response_store.put(response_id, {
+"""
+
+RUNS_AFTER = """        run_id = f"run_{uuid.uuid4().hex}"
         session_id = body.get("session_id") or stored_session_id or run_id
 
         # Persist API-run lineage before the agent starts. Automated callers
@@ -183,13 +208,13 @@ RUNS_AFTER = '''        run_id = f"run_{uuid.uuid4().hex}"
                 db.reopen_session(session_id)
 
         # Approval queues gate host-side tool execution and must be isolated
-'''
+"""
 
-RUN_CLOSE_BEFORE = '''            finally:
+RUN_CLOSE_BEFORE = """            finally:
                 # If the asyncio wrapper is cancelled (for example via
-'''
+"""
 
-RUN_CLOSE_AFTER = '''            finally:
+RUN_CLOSE_AFTER = """            finally:
                 # Parent-linked API workers are durable dashboard sessions.
                 # Close them on every terminal run path so the dashboard can
                 # distinguish a long model wait from completed or failed work.
@@ -218,10 +243,10 @@ RUN_CLOSE_AFTER = '''            finally:
                         )
 
                 # If the asyncio wrapper is cancelled (for example via
-'''
+"""
 
-EVENT_CALLBACK_SIGNATURE_BEFORE = '''    def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop"):
+EVENT_CALLBACK_SIGNATURE_BEFORE = """    def _make_run_event_callback(self, run_id: str, loop: "asyncio.AbstractEventLoop"):
-'''
+"""
 
 EVENT_CALLBACK_SIGNATURE_AFTER = '''    _RUN_ACTIVITY_BYTES = 512_000
     _RUN_ACTIVITY_FILES = 256
@@ -346,12 +371,12 @@ EVENT_CALLBACK_SIGNATURE_AFTER = '''    _RUN_ACTIVITY_BYTES = 512_000
     ):
 '''
 
-EVENT_CALLBACK_BODY_BEFORE = '''        def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
+EVENT_CALLBACK_BODY_BEFORE = """        def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
             ts = time.time()
             if event_type == "tool.started":
-'''
+"""
 
-EVENT_CALLBACK_BODY_AFTER = '''        def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
+EVENT_CALLBACK_BODY_AFTER = """        def _callback(event_type: str, tool_name: str = None, preview: str = None, args=None, **kwargs):
             ts = time.time()
             self._record_run_activity(
                 session_id,
@@ -361,12 +386,12 @@ EVENT_CALLBACK_BODY_AFTER = '''        def _callback(event_type: str, tool_name:
                 is_error=bool(kwargs.get("is_error", False)),
             )
             if event_type == "tool.started":
-'''
+"""
 
-EVENT_CALLBACK_END_BEFORE = '''            # _thinking and subagent_progress are intentionally not forwarded
+EVENT_CALLBACK_END_BEFORE = """            # _thinking and subagent_progress are intentionally not forwarded
-'''
+"""
 
-EVENT_CALLBACK_END_AFTER = '''            elif event_type in {
+EVENT_CALLBACK_END_AFTER = """            elif event_type in {
                 "subagent.tool",
                 "subagent.progress",
                 "subagent_progress",
@@ -386,27 +411,27 @@ EVENT_CALLBACK_END_AFTER = '''            elif event_type in {
                     "tool": tool_name if event_type == "subagent.tool" else None,
                     "preview": safe_preview,
                 })
-'''
+"""
 
-EVENT_CALLBACK_CALL_BEFORE = '''        event_cb = self._make_run_event_callback(run_id, loop)
+EVENT_CALLBACK_CALL_BEFORE = """        event_cb = self._make_run_event_callback(run_id, loop)
-'''
+"""
 
-EVENT_CALLBACK_CALL_AFTER = '''        event_cb = self._make_run_event_callback(
+EVENT_CALLBACK_CALL_AFTER = """        event_cb = self._make_run_event_callback(
             run_id,
             loop,
             session_id=session_id,
         )
         self._record_run_activity(session_id, "run.started")
-'''
+"""
 
-RUN_SWEEP_BEFORE = '''                self._run_streams.pop(run_id, None)
+RUN_SWEEP_BEFORE = """                self._run_streams.pop(run_id, None)
                 self._run_streams_created.pop(run_id, None)
                 self._active_run_agents.pop(run_id, None)
                 self._active_run_tasks.pop(run_id, None)
                 self._run_approval_sessions.pop(run_id, None)
-'''
+"""
 
-RUN_SWEEP_AFTER = '''                self._run_streams.pop(run_id, None)
+RUN_SWEEP_AFTER = """                self._run_streams.pop(run_id, None)
                 self._run_streams_created.pop(run_id, None)
                 # Stream retention and run lifetime are separate. A long run
                 # can legitimately outlive its unconsumed SSE queue; keep the
@@ -416,7 +441,7 @@ RUN_SWEEP_AFTER = '''                self._run_streams.pop(run_id, None)
                     self._active_run_agents.pop(run_id, None)
                     self._active_run_tasks.pop(run_id, None)
                     self._run_approval_sessions.pop(run_id, None)
-'''
+"""
 
 
 def patch(source: Path, destination: Path) -> None:
@@ -444,6 +469,9 @@ def patch(source: Path, destination: Path) -> None:
     content = content.replace(RUNS_BEFORE, RUNS_AFTER, 1)
     content = content.replace(RUN_CLOSE_BEFORE, RUN_CLOSE_AFTER, 1)
     content = content.replace(RESPONSES_SESSION_BEFORE, RESPONSES_SESSION_AFTER, 1)
+    content = content.replace(TRUNCATION_BEFORE, TRUNCATION_AFTER, 1)
+    content = content.replace(HISTORY_STORE_BEFORE, HISTORY_STORE_AFTER, 1)
+    content = content.replace(SSE_STORE_BEFORE, SSE_STORE_AFTER, 1)
     content = content.replace(
         EVENT_CALLBACK_SIGNATURE_BEFORE,
         EVENT_CALLBACK_SIGNATURE_AFTER,

testing/tests/test_hermes_chat_smoothness.py

@@ -0,0 +1,493 @@
+"""Adversarial contracts for chat-only continuity and truthful rendering."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import runpy
+import sqlite3
+import sys
+from pathlib import Path
+
+import pytest
+import yaml
+
+
+ROOT = Path(__file__).parents[2]
+HERMES = ROOT / "services" / "hermes"
+
+
+def _load(name: str, path: Path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture
+def continuity():
+    return _load(
+        "telegram_continuity",
+        HERMES / "scripts" / "migrate_telegram_api_sessions.py",
+    )
+
+
+def test_context_is_deterministic_bounded_deduplicated_and_truthful(continuity):
+    turns = []
+    for index in range(30):
+        user = {"role": "user", "content": f"question {index}", "_db_persisted": True}
+        assistant = {
+            "role": "assistant",
+            "content": f"answer {index}",
+            "_db_persisted": True,
+        }
+        turns.extend((user, assistant))
+        if index == 4:
+            # Simulate one complete transcript replay.
+            turns.extend((user.copy(), assistant.copy()))
+    turns.extend(
+        [
+            {
+                "role": "assistant",
+                "content": "",
+                "tool_calls": [
+                    {
+                        "id": "call-1",
+                        "type": "function",
+                        "function": {
+                            "name": "browser",
+                            "arguments": "x" * 40_000,
+                        },
+                    }
+                ],
+            },
+            {
+                "role": "tool",
+                "tool_call_id": "call-1",
+                "tool_name": "browser",
+                "content": "tool output " * 4_000,
+            },
+            {
+                "role": "user",
+                "content": [
+                    {"type": "input_text", "text": "edit this"},
+                    {
+                        "type": "input_image",
+                        "image_url": "data:image/png;base64," + "a" * 80_000,
+                    },
+                ],
+            },
+            {
+                "role": "assistant",
+                "content": "Created it MEDIA:/opt/data/cache/images/private.png",
+            },
+        ]
+    )
+
+    compacted = continuity.compact_telegram_history(turns)
+
+    assert compacted == continuity.compact_telegram_history(turns)
+    assert compacted[0]["role"] == "system"
+    assert compacted[0]["_db_persisted"] is True
+    assert compacted[0]["content"].startswith(continuity.SUMMARY_PREFIX)
+    assert sum(item["role"] == "user" for item in compacted[1:]) <= 12
+    assert len(compacted[1:]) <= continuity.MAX_HISTORY_ITEMS
+    assert len(json.dumps(compacted).encode()) <= (
+        continuity.MAX_HISTORY_BYTES + continuity.MAX_SUMMARY_CHARS
+    )
+    encoded = json.dumps(compacted)
+    assert "data:image" not in encoded
+    assert "MEDIA:" not in encoded
+    assert "/opt/data/" not in encoded
+    assert continuity.ATTACHMENT_NOTE in encoded
+    assert "question 29" in encoded
+    assert (
+        next(item for item in compacted if item.get("content") == "question 29")[
+            "_db_persisted"
+        ]
+        is True
+    )
+    assert "x" * 3_000 not in encoded
+    fingerprints = [continuity._fingerprint(item) for item in compacted]
+    assert all(
+        left != right
+        for left, right in zip(fingerprints, fingerprints[1:], strict=False)
+    )
+
+
+def test_context_rejects_malformed_history_and_bounds_single_large_item(continuity):
+    assert continuity.compact_telegram_history(None) == []
+    assert continuity.compact_telegram_history({"role": "user"}) == []
+    malformed = [
+        None,
+        "text",
+        {"role": "unknown", "content": "skip"},
+        {"role": "tool", "content": ""},
+        {"role": "user", "content": {"nested": "value"}},
+        {"role": "assistant", "content": "z" * 2_000_000},
+    ]
+    compacted = continuity.compact_telegram_history(malformed)
+    assert len(compacted) == 2
+    assert compacted[0]["content"] == '{"nested": "value"}'
+    assert len(compacted[1]["content"]) == continuity.MAX_ITEM_CHARS
+
+
+def test_context_defensive_normalization_and_replay_edges(continuity, monkeypatch):
+    assert continuity._text(["skip", {"type": "text", "text": "keep"}]) == "keep"
+    assert continuity._text({"not-json-serializable"}) == "{'not-json-serializable'}"
+    normalized = continuity._normalized_item(
+        {
+            "role": "assistant",
+            "tool_calls": [
+                None,
+                {"id": "call", "function": {"name": "browser", "arguments": {1}}},
+            ],
+        }
+    )
+    assert normalized and normalized["tool_calls"][0]["id"] == "call"
+    assert continuity._fingerprint({"role": "user", "content": {1}}).startswith(
+        "{'role':"
+    )
+
+    user = {"role": "user", "content": "same"}
+    assistant = {"role": "assistant", "content": "reply"}
+    assert continuity._deduplicate(
+        [user, assistant, user.copy(), assistant.copy()]
+    ) == [
+        user,
+        assistant,
+    ]
+    assert continuity._deduplicate(
+        [
+            {"role": "tool", "content": "first", "id": "stable"},
+            {"role": "tool", "content": "replay", "id": "stable"},
+        ]
+    ) == [{"role": "tool", "content": "first", "id": "stable"}]
+
+    summary = continuity._summary(
+        [
+            {
+                "role": "system",
+                "content": continuity.SUMMARY_PREFIX + "\n- existing fact",
+            },
+            {"role": "tool", "content": "not durable prose"},
+        ]
+    )
+    assert summary and "existing fact" in summary["content"]
+    monkeypatch.setattr(
+        continuity, "MAX_SUMMARY_CHARS", len(continuity.SUMMARY_PREFIX) + 1
+    )
+    assert (
+        continuity._summary([{"role": "user", "content": "too long"}])["content"]
+        == continuity.SUMMARY_PREFIX
+    )
+
+
+def test_context_final_serialized_size_guard(continuity, monkeypatch):
+    monkeypatch.setattr(continuity, "MAX_HISTORY_ITEMS", 2)
+    monkeypatch.setattr(continuity, "MAX_HISTORY_BYTES", 1_000)
+    monkeypatch.setattr(
+        continuity,
+        "_summary",
+        lambda _older: {
+            "role": "system",
+            "content": "s" * 20_000,
+            "_db_persisted": True,
+        },
+    )
+    compacted = continuity.compact_telegram_history(
+        [
+            {"role": "user", "content": "older"},
+            {"role": "assistant", "content": "recent one"},
+            {"role": "assistant", "content": "recent two"},
+        ]
+    )
+    assert [item["content"] for item in compacted[1:]] == ["recent two"]
+
+
+def _create_session_databases(tmp_path: Path):
+    state = tmp_path / "state.db"
+    responses = tmp_path / "response_store.db"
+    with sqlite3.connect(state) as connection:
+        connection.execute(
+            """CREATE TABLE sessions (
+                id TEXT PRIMARY KEY, source TEXT, session_key TEXT, chat_type TEXT,
+                display_name TEXT, origin_json TEXT, title TEXT
+            )"""
+        )
+        connection.executemany(
+            "INSERT INTO sessions VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (
+                (
+                    "legacy",
+                    "api_server",
+                    "wrong",
+                    "group",
+                    "Unassigned",
+                    "{}",
+                    "Unassigned",
+                ),
+                ("named", "api_server", None, None, None, None, "Keep this title"),
+                ("other", "cli", None, None, "Unassigned", None, None),
+            ),
+        )
+        connection.execute(
+            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, timestamp REAL)"
+        )
+        connection.executemany(
+            "INSERT INTO messages VALUES (?, 'legacy', 'user', 'same', 1)", ((1,), (2,))
+        )
+    with sqlite3.connect(responses) as connection:
+        connection.execute(
+            "CREATE TABLE conversations (name TEXT PRIMARY KEY, response_id TEXT NOT NULL)"
+        )
+        connection.execute(
+            "CREATE TABLE responses (response_id TEXT PRIMARY KEY, data TEXT NOT NULL, accessed_at REAL NOT NULL)"
+        )
+        connection.executemany(
+            "INSERT INTO responses VALUES (?, ?, 0)",
+            (
+                ("general", json.dumps({"session_id": "legacy"})),
+                ("topic", json.dumps({"session_id": "named"})),
+                ("bad-json", "{"),
+                ("bad-id", json.dumps({"session_id": "../unsafe"})),
+                (
+                    "oversized",
+                    json.dumps({"session_id": "other", "padding": "x" * 500}),
+                ),
+            ),
+        )
+        connection.executemany(
+            "INSERT INTO conversations VALUES (?, ?)",
+            (
+                ("telegram", "general"),
+                ("telegram-topic-plans-g1", "topic"),
+                ("telegram-topic-malformed", "bad-json"),
+                ("telegram-topic-unsafe", "bad-id"),
+                ("telegram-topic-oversized", "oversized"),
+            ),
+        )
+    return state, responses
+
+
+def test_existing_session_migration_repairs_labels_and_is_idempotent(
+    tmp_path: Path, continuity, monkeypatch
+):
+    state, responses = _create_session_databases(tmp_path)
+    monkeypatch.setattr(continuity, "MAX_RESPONSE_BYTES", 300)
+
+    assert continuity.migrate(state, responses) == 2
+    assert continuity.migrate(state, responses) == 0
+
+    with sqlite3.connect(state) as connection:
+        rows = {
+            row[0]: row[1:]
+            for row in connection.execute(
+                "SELECT id, session_key, chat_type, display_name, origin_json, title "
+                "FROM sessions ORDER BY id"
+            )
+        }
+    assert rows["legacy"][:3] == ("telegram", "private", "Telegram")
+    assert json.loads(rows["legacy"][3]) == {
+        "platform": "telegram",
+        "session_key": "telegram",
+    }
+    assert rows["legacy"][4] == "Telegram · General"
+    assert rows["named"][:3] == (
+        "telegram-topic-plans-g1",
+        "private",
+        "Telegram",
+    )
+    assert rows["named"][4] == "Keep this title"
+    assert rows["other"] == (None, None, "Unassigned", None, None)
+    with sqlite3.connect(state) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM messages").fetchone() == (1,)
+
+
+def test_migration_fails_closed_on_missing_or_malformed_databases(
+    tmp_path: Path, continuity
+):
+    assert continuity.telegram_sessions(tmp_path / "missing.db") == {}
+    assert (
+        continuity.migrate(tmp_path / "missing-state.db", tmp_path / "missing.db") == 0
+    )
+    broken = tmp_path / "broken.db"
+    broken.write_text("not sqlite", encoding="utf-8")
+    assert continuity.telegram_sessions(broken) == {}
+
+    state = tmp_path / "state.db"
+    responses = tmp_path / "responses.db"
+    with sqlite3.connect(state) as connection:
+        connection.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY)")
+    with sqlite3.connect(responses) as connection:
+        connection.execute("CREATE TABLE conversations (name TEXT, response_id TEXT)")
+        connection.execute("CREATE TABLE responses (response_id TEXT, data TEXT)")
+        connection.execute(
+            "INSERT INTO responses VALUES ('valid', '{\"session_id\": \"known\"}')"
+        )
+        connection.execute("INSERT INTO conversations VALUES ('telegram', 'valid')")
+    assert continuity.migrate(state, responses) == 0
+
+
+def test_migration_skips_invalid_conversation_and_rolls_back_database_errors(
+    tmp_path: Path, continuity
+):
+    state = tmp_path / "state.db"
+    responses = tmp_path / "responses.db"
+    with sqlite3.connect(state) as connection:
+        connection.execute(
+            """CREATE TABLE sessions (
+                id TEXT PRIMARY KEY, source TEXT, session_key TEXT, chat_type TEXT,
+                display_name TEXT, origin_json TEXT, title TEXT
+            )"""
+        )
+        connection.execute(
+            "INSERT INTO sessions VALUES ('known', 'api_server', NULL, NULL, NULL, NULL, NULL)"
+        )
+        connection.execute(
+            """CREATE TRIGGER reject_updates BEFORE UPDATE ON sessions
+               BEGIN SELECT RAISE(FAIL, 'read only for test'); END"""
+        )
+    with sqlite3.connect(responses) as connection:
+        connection.execute(
+            "CREATE TABLE conversations (name TEXT PRIMARY KEY, response_id TEXT NOT NULL)"
+        )
+        connection.execute(
+            "CREATE TABLE responses (response_id TEXT PRIMARY KEY, data TEXT NOT NULL)"
+        )
+        connection.executemany(
+            "INSERT INTO responses VALUES (?, ?)",
+            (
+                ("valid", json.dumps({"session_id": "known"})),
+                ("invalid", json.dumps({"session_id": "ignored"})),
+            ),
+        )
+        connection.executemany(
+            "INSERT INTO conversations VALUES (?, ?)",
+            (("telegram", "valid"), ("telegram-topic-BAD", "invalid")),
+        )
+
+    assert continuity.telegram_sessions(responses) == {"known": "telegram"}
+    assert continuity.migrate(state, responses) == 0
+    with sqlite3.connect(state) as connection:
+        assert connection.execute(
+            "SELECT session_key FROM sessions WHERE id = 'known'"
+        ).fetchone() == (None,)
+
+
+def test_api_patch_compacts_input_batch_and_disconnect_snapshots(tmp_path: Path):
+    patcher = _load(
+        "chat_api_patch",
+        HERMES / "scripts" / "patch_api_server_sessions.py",
+    )
+    source = tmp_path / "api_server.py"
+    destination = tmp_path / "patched.py"
+    source.write_text(
+        patcher.BEFORE
+        + patcher.RUNS_BEFORE
+        + patcher.RUN_CLOSE_BEFORE
+        + patcher.TRUNCATION_BEFORE
+        + patcher.RESPONSES_SESSION_BEFORE
+        + patcher.EVENT_CALLBACK_SIGNATURE_BEFORE
+        + patcher.EVENT_CALLBACK_BODY_BEFORE
+        + patcher.EVENT_CALLBACK_END_BEFORE
+        + patcher.EVENT_CALLBACK_CALL_BEFORE
+        + patcher.RUN_SWEEP_BEFORE
+        + patcher.SSE_STORE_BEFORE
+        + patcher.HISTORY_STORE_BEFORE,
+        encoding="utf-8",
+    )
+
+    patcher.patch(source, destination)
+    patched = destination.read_text(encoding="utf-8")
+
+    assert (
+        "conversation_history = compact_telegram_history(conversation_history)"
+        in patched
+    )
+    assert "conversation_history_snapshot = compact_telegram_history(" in patched
+    assert "full_history = compact_telegram_history(full_history)" in patched
+    assert patched.count("from gateway.platforms.telegram_continuity import") == 2
+
+
+def _patch_source(patcher) -> str:
+    return (
+        patcher.BEFORE
+        + patcher.RUNS_BEFORE
+        + patcher.RUN_CLOSE_BEFORE
+        + patcher.RESPONSES_SESSION_BEFORE
+        + patcher.EVENT_CALLBACK_SIGNATURE_BEFORE
+        + patcher.EVENT_CALLBACK_BODY_BEFORE
+        + patcher.EVENT_CALLBACK_END_BEFORE
+        + patcher.EVENT_CALLBACK_CALL_BEFORE
+        + patcher.RUN_SWEEP_BEFORE
+    )
+
+
+@pytest.mark.parametrize(
+    "marker_name",
+    [
+        "BEFORE",
+        "RUNS_BEFORE",
+        "RUN_CLOSE_BEFORE",
+        "RESPONSES_SESSION_BEFORE",
+        "EVENT_CALLBACK_SIGNATURE_BEFORE",
+    ],
+)
+def test_api_patch_fails_closed_on_upstream_drift(tmp_path: Path, marker_name: str):
+    patcher = _load(
+        "chat_api_patch_drift",
+        HERMES / "scripts" / "patch_api_server_sessions.py",
+    )
+    source = tmp_path / "api_server.py"
+    marker = getattr(patcher, marker_name)
+    source.write_text(_patch_source(patcher).replace(marker, "", 1), encoding="utf-8")
+    with pytest.raises(RuntimeError, match="patch context changed"):
+        patcher.patch(source, tmp_path / "patched.py")
+
+
+def test_script_entrypoints(tmp_path: Path, monkeypatch):
+    continuity_path = HERMES / "scripts" / "migrate_telegram_api_sessions.py"
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            str(continuity_path),
+            str(tmp_path / "state.db"),
+            str(tmp_path / "responses.db"),
+        ],
+    )
+    with pytest.raises(SystemExit, match="0"):
+        runpy.run_path(str(continuity_path), run_name="__main__")
+
+    patch_path = HERMES / "scripts" / "patch_api_server_sessions.py"
+    patcher = _load("chat_api_patch_cli", patch_path)
+    source = tmp_path / "api_server.py"
+    destination = tmp_path / "patched.py"
+    source.write_text(_patch_source(patcher), encoding="utf-8")
+    monkeypatch.setattr(sys, "argv", [str(patch_path), str(source), str(destination)])
+    with pytest.raises(SystemExit, match="0"):
+        runpy.run_path(str(patch_path), run_name="__main__")
+    assert destination.exists()
+
+
+def test_chat_mounts_continuity_only_into_isolated_tenants():
+    statefulset = yaml.safe_load((HERMES / "chat-statefulset.yaml").read_text())
+    pod = statefulset["spec"]["template"]["spec"]
+    hermes = next(item for item in pod["containers"] if item["name"] == "hermes")
+    assert {
+        "name": "coordinator",
+        "mountPath": "/opt/hermes/gateway/platforms/telegram_continuity.py",
+        "subPath": "migrate_telegram_api_sessions.py",
+        "readOnly": True,
+    } in hermes["volumeMounts"]
+    patch_init = next(
+        item
+        for item in pod["initContainers"]
+        if item["name"] == "patch-api-server-sessions"
+    )
+    command = patch_init["args"][0]
+    assert command.count("grep -Fq") == 3
+    assert "compact_telegram_history" in command
+    assert "agent-deployment" not in (HERMES / "chat-statefulset.yaml").read_text()