From c867ae52fbb8d424053c711938d2fa7a36e9fe06 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Mon, 17 Aug 2026 14:39:21 +0000 Subject: [PATCH] hermes: harden chat continuity and activity --- services/hermes/chat-statefulset.yaml | 4 + services/hermes/router/main.go | 12 +- services/hermes/router/main_test.go | 68 +++ services/hermes/router/session_continuity.go | 97 ++++ services/hermes/router/session_snapshot.go | 65 +++ .../hermes/router/session_snapshot_test.go | 113 ++++ services/hermes/router/telegram.go | 68 ++- services/hermes/router/telegram_input.go | 88 ++++ services/hermes/router/telegram_media.go | 8 +- services/hermes/router/telegram_media_test.go | 14 + services/hermes/router/telegram_test.go | 139 +++++ services/hermes/router/telegram_topics.go | 55 +- services/hermes/router/web.go | 12 + .../scripts/migrate_telegram_api_sessions.py | 328 +++++++++++- .../scripts/patch_api_server_sessions.py | 112 ++-- testing/tests/test_hermes_chat_smoothness.py | 493 ++++++++++++++++++ 16 files changed, 1564 insertions(+), 112 deletions(-) create mode 100644 services/hermes/router/session_continuity.go create mode 100644 services/hermes/router/session_snapshot.go create mode 100644 services/hermes/router/session_snapshot_test.go create mode 100644 testing/tests/test_hermes_chat_smoothness.py diff --git a/services/hermes/chat-statefulset.yaml b/services/hermes/chat-statefulset.yaml index dda35cd7..3d509116 100644 --- a/services/hermes/chat-statefulset.yaml +++ b/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} diff --git a/services/hermes/router/main.go b/services/hermes/router/main.go index 16d7656c..db115242 100644 --- a/services/hermes/router/main.go +++ b/services/hermes/router/main.go @@ -27,8 +27,9 @@ type linkRecord struct { } type telegramTopic struct { - Label string `json:"label"` - LastUsed int64 `json:"last_used"` + Label string `json:"label"` + 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) diff --git a/services/hermes/router/main_test.go b/services/hermes/router/main_test.go index a17b020a..1e402bec 100644 --- a/services/hermes/router/main_test.go +++ b/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 { diff --git a/services/hermes/router/session_continuity.go b/services/hermes/router/session_continuity.go new file mode 100644 index 00000000..af42b35d --- /dev/null +++ b/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); +})();` diff --git a/services/hermes/router/session_snapshot.go b/services/hermes/router/session_snapshot.go new file mode 100644 index 00000000..237dd19b --- /dev/null +++ b/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 +} diff --git a/services/hermes/router/session_snapshot_test.go b/services/hermes/router/session_snapshot_test.go new file mode 100644 index 00000000..64aee010 --- /dev/null +++ b/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) + } +} diff --git a/services/hermes/router/telegram.go b/services/hermes/router/telegram.go index c4cc3b66..a2d591df 100644 --- a/services/hermes/router/telegram.go +++ b/services/hermes/router/telegram.go @@ -24,20 +24,21 @@ type telegramConfig struct { } type telegramBot struct { - config telegramConfig - router *tenantRouter - apiBase string - fileBase string - client *http.Client - mediaClient *http.Client - agentClient *http.Client - mu sync.RWMutex - botUsername string - ready bool - lastError string - workLimit chan struct{} - slotLocks map[int]*sync.Mutex - slotLocksMux sync.Mutex + config telegramConfig + router *tenantRouter + apiBase string + fileBase string + client *http.Client + mediaClient *http.Client + agentClient *http.Client + mu sync.RWMutex + botUsername string + ready bool + lastError string + workLimit chan struct{} + activityEvery time.Duration + 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, - router: router, - apiBase: "https://api.telegram.org/bot" + config.BotToken, - fileBase: "https://api.telegram.org/file/bot" + config.BotToken, - client: &http.Client{Timeout: 70 * time.Second}, - mediaClient: newTenantMediaClient(), - agentClient: &http.Client{Timeout: 15 * time.Minute}, - workLimit: make(chan struct{}, 4), - slotLocks: map[int]*sync.Mutex{}, + config: config, + router: router, + apiBase: "https://api.telegram.org/bot" + config.BotToken, + fileBase: "https://api.telegram.org/file/bot" + config.BotToken, + client: &http.Client{Timeout: 70 * time.Second}, + mediaClient: newTenantMediaClient(), + agentClient: &http.Client{Timeout: 15 * time.Minute}, + workLimit: make(chan struct{}, 4), + 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 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 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}, diff --git a/services/hermes/router/telegram_input.go b/services/hermes/router/telegram_input.go index 40be7f96..5e2c5956 100644 --- a/services/hermes/router/telegram_input.go +++ b/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 { diff --git a/services/hermes/router/telegram_media.go b/services/hermes/router/telegram_media.go index 90be976f..d4dacca9 100644 --- a/services/hermes/router/telegram_media.go +++ b/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 diff --git a/services/hermes/router/telegram_media_test.go b/services/hermes/router/telegram_media_test.go index ee36f3d0..ff59bb2c 100644 --- a/services/hermes/router/telegram_media_test.go +++ b/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", diff --git a/services/hermes/router/telegram_test.go b/services/hermes/router/telegram_test.go index 317dcadc..b192ad14 100644 --- a/services/hermes/router/telegram_test.go +++ b/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 { diff --git a/services/hermes/router/telegram_topics.go b/services/hermes/router/telegram_topics.go index d5e9bed6..59c1447a 100644 --- a/services/hermes/router/telegram_topics.go +++ b/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 { diff --git a/services/hermes/router/web.go b/services/hermes/router/web.go index 0ea02a20..4536eb12 100644 --- a/services/hermes/router/web.go +++ b/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, "", ``, 1) content = strings.Replace(content, "", ``, 1) } + if !strings.Contains(content, "hermes-session-continuity.js") { + content = strings.Replace(content, "", ``, 1) + } response.Body = io.NopCloser(strings.NewReader(content)) response.ContentLength = int64(len(content)) response.Header.Set("Content-Length", strconv.Itoa(len(content))) diff --git a/services/hermes/scripts/migrate_telegram_api_sessions.py b/services/hermes/scripts/migrate_telegram_api_sessions.py index 82f58456..9d4597f9 100644 --- a/services/hermes/scripts/migrate_telegram_api_sessions.py +++ b/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: - session_id = str(json.loads(raw).get("session_id") or "").strip() - except (TypeError, ValueError, json.JSONDecodeError): + name = str(conversation or "") + if not _CONVERSATION.fullmatch(name): continue - if session_id: - result[session_id] = str(conversation) + try: + 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]) - for row in connection.execute("PRAGMA table_info(sessions)").fetchall() + str(row[1]) for row in connection.execute("PRAGMA table_info(sessions)") } required = { - "id", "source", "session_key", "chat_type", "display_name", - "origin_json", "title", + "id", + "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, ?), - chat_type = COALESCE(chat_type, 'private'), - display_name = COALESCE(display_name, 'Telegram'), - origin_json = COALESCE(origin_json, ?), - title = COALESCE(title, ?) + SET session_key = ?, chat_type = 'private', + display_name = 'Telegram', origin_json = ?, + title = CASE + WHEN title IS NULL OR TRIM(title) = '' OR + LOWER(TRIM(title)) = 'unassigned' + THEN ? ELSE title END WHERE id = ? AND source = 'api_server' - AND ( - session_key IS NULL OR chat_type IS NULL OR - display_name IS NULL OR origin_json IS NULL OR title IS NULL - )""", - (conversation, origin, title, session_id), + AND (COALESCE(session_key, '') != ? OR + COALESCE(chat_type, '') != 'private' OR + COALESCE(display_name, '') != 'Telegram' OR + COALESCE(origin_json, '') != ? OR title IS NULL OR + 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 diff --git a/services/hermes/scripts/patch_api_server_sessions.py b/services/hermes/scripts/patch_api_server_sessions.py index bc7cafcc..55a57b81 100644 --- a/services/hermes/scripts/patch_api_server_sessions.py +++ b/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", "" - ).strip().lower() + # X-Hermes-Conversation-Platform was parsed before history compaction. 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, diff --git a/testing/tests/test_hermes_chat_smoothness.py b/testing/tests/test_hermes_chat_smoothness.py new file mode 100644 index 00000000..ee7a32eb --- /dev/null +++ b/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()