From 0dd6ea0f02672ecafd8c5c3bde7f5677e960c164 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 15:59:09 -0300 Subject: [PATCH] hermes: add durable Telegram topics --- dockerfiles/Dockerfile.hermes-webui | 3 + .../hermes-webui-telegram-project-patch.py | 114 ++++++++++++++ services/hermes/chat-router.yaml | 2 +- services/hermes/chat-statefulset.yaml | 30 +++- services/hermes/deployment.yaml | 2 +- services/hermes/kustomization.yaml | 1 + services/hermes/router/main.go | 40 +++-- services/hermes/router/telegram.go | 53 +++++-- services/hermes/router/telegram_input.go | 14 +- services/hermes/router/telegram_test.go | 62 +++++++- services/hermes/router/telegram_topics.go | 139 ++++++++++++++++++ .../scripts/migrate_telegram_api_sessions.py | 92 ++++++++++++ .../scripts/patch_api_server_sessions.py | 69 +++++++++ testing/tests/test_hermes_chat_quality.py | 86 ++++++++++- 14 files changed, 673 insertions(+), 34 deletions(-) create mode 100644 dockerfiles/hermes-webui-telegram-project-patch.py create mode 100644 services/hermes/router/telegram_topics.go create mode 100644 services/hermes/scripts/migrate_telegram_api_sessions.py diff --git a/dockerfiles/Dockerfile.hermes-webui b/dockerfiles/Dockerfile.hermes-webui index 1280c8d9..a9f22721 100644 --- a/dockerfiles/Dockerfile.hermes-webui +++ b/dockerfiles/Dockerfile.hermes-webui @@ -90,10 +90,12 @@ PY # Add the Atlas voice bridge as a narrow integration layer. It activates only # when a tenant's server-side STT capability reports the private Jetson route. COPY dockerfiles/hermes-webui-atlas-patch.py /tmp/hermes-webui-atlas-patch.py +COPY dockerfiles/hermes-webui-telegram-project-patch.py /tmp/hermes-webui-telegram-project-patch.py COPY dockerfiles/hermes-webui-atlas-voice.js /opt/hermes-webui/static/atlas-voice.js COPY dockerfiles/hermes-webui-router-patch.py /tmp/hermes-webui-router-patch.py COPY dockerfiles/hermes-webui-router.js /opt/hermes-webui/static/atlas-router.js RUN /opt/hermes/.venv/bin/python /tmp/hermes-webui-atlas-patch.py +RUN /opt/hermes/.venv/bin/python /tmp/hermes-webui-telegram-project-patch.py RUN /opt/hermes/.venv/bin/python /tmp/hermes-webui-router-patch.py RUN /opt/hermes/.venv/bin/python -c 'import cryptography, yaml' \ @@ -102,6 +104,7 @@ RUN /opt/hermes/.venv/bin/python -c 'import cryptography, yaml' \ && ! grep -Fq 'data-effort="max"' /opt/hermes-webui/static/index.html \ && grep -Fq "window.location.assign('/oauth2/start?rd='" /opt/hermes-webui/static/ui.js \ && grep -Fq "childrenExpanded?'▾ ':'▸ '" /opt/hermes-webui/static/sessions.js \ + && grep -Fq "TELEGRAM_PROJECT_NAME = 'Telegram'" /opt/hermes-webui/api/models.py \ && grep -Fq "'atlas/auto/maximum': 'Automatic · Maximum'" /opt/hermes-webui/static/panels.js \ && grep -Fq 'Atlas Jetson (private)' /opt/hermes-webui/static/index.html \ && grep -Fq 'HERMES_WEBUI_ATLAS_TTS_URL' /opt/hermes-webui/api/routes.py \ diff --git a/dockerfiles/hermes-webui-telegram-project-patch.py b/dockerfiles/hermes-webui-telegram-project-patch.py new file mode 100644 index 00000000..e1fe9c38 --- /dev/null +++ b/dockerfiles/hermes-webui-telegram-project-patch.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Group trusted Telegram API sessions under a system WebUI project.""" + +from __future__ import annotations + +from pathlib import Path + + +ROOT = Path("/opt/hermes-webui") + + +def replace_exact(path: Path, before: str, after: str, count: int = 1) -> None: + """Replace a pinned upstream fragment and fail closed on source drift.""" + source = path.read_text(encoding="utf-8") + if source.count(before) != count: + raise SystemExit(f"Telegram project patch context changed in {path}") + path.write_text(source.replace(before, after, count), encoding="utf-8") + + +models = ROOT / "api/models.py" + +replace_exact( + models, + """def _profile_has_user_projects() -> bool: +""", + """TELEGRAM_PROJECT_NAME = 'Telegram' +_TELEGRAM_PROJECT_LOCK = threading.Lock() + + +def ensure_telegram_project() -> str: + \"\"\"Return the per-profile system project for trusted Telegram sessions.\"\"\" + from api.profiles import get_active_profile_name, _is_root_profile + + active = get_active_profile_name() or 'default' + with _TELEGRAM_PROJECT_LOCK: + projects = load_projects() + for project in projects: + if project.get('name') != TELEGRAM_PROJECT_NAME: + continue + row_profile = project.get('profile') + if row_profile == active: + return project['project_id'] + if _is_root_profile(row_profile or 'default') and _is_root_profile(active): + return project['project_id'] + project_id = uuid.uuid4().hex[:12] + projects.append({ + 'project_id': project_id, + 'name': TELEGRAM_PROJECT_NAME, + 'color': '#229ed9', + 'profile': active, + 'created_at': time.time(), + }) + save_projects(projects) + return project_id + + +def _profile_has_user_projects() -> bool: +""", +) + +replace_exact( + models, + "reserved = {CRON_PROJECT_NAME, WEBHOOK_PROJECT_NAME}", + "reserved = {CRON_PROJECT_NAME, WEBHOOK_PROJECT_NAME, TELEGRAM_PROJECT_NAME}", +) + +replace_exact( + models, + """ _webhook_pid_cache: list[str | None] = [None] + def _webhook_pid(): + if _webhook_pid_cache[0] is None: + _webhook_pid_cache[0] = ensure_webhook_project() + return _webhook_pid_cache[0] + + def _state_row_project_id(sid: str, source: str | None) -> str | None: + if is_cron_session(sid, source): + return _cron_pid() + if is_webhook_session(sid, source): + return _webhook_pid() + return None +""", + """ _webhook_pid_cache: list[str | None] = [None] + def _webhook_pid(): + if _webhook_pid_cache[0] is None: + _webhook_pid_cache[0] = ensure_webhook_project() + return _webhook_pid_cache[0] + + _telegram_pid_cache: list[str | None] = [None] + def _telegram_pid(): + if _telegram_pid_cache[0] is None: + _telegram_pid_cache[0] = ensure_telegram_project() + return _telegram_pid_cache[0] + + def _state_row_project_id(row: dict) -> str | None: + sid = str(row.get('id') or '') + source = str(row.get('source') or '') + if is_cron_session(sid, source): + return _cron_pid() + if is_webhook_session(sid, source): + return _webhook_pid() + session_key = str(row.get('session_key') or '') + if source == 'api_server' and ( + session_key == 'telegram' or session_key.startswith('telegram-topic-') + ): + return _telegram_pid() + return None +""", +) + +replace_exact( + models, + "'project_id': _state_row_project_id(sid, _source),", + "'project_id': _state_row_project_id(row),", +) diff --git a/services/hermes/chat-router.yaml b/services/hermes/chat-router.yaml index 142c5112..241e82a2 100644 --- a/services/hermes/chat-router.yaml +++ b/services/hermes/chat-router.yaml @@ -62,7 +62,7 @@ spec: values: [rpi5] containers: - name: router - image: registry.bstein.dev/bstein/hermes-chat-router@sha256:cec7d91b6aed26c2272e6caae731286cfd4122e35e818ea06b1c19d7434f0520 + image: registry.bstein.dev/bstein/hermes-chat-router@sha256:89881f38317c5fdced6e995755a7e52d5e483a43d90e278341e564a799ecc74b imagePullPolicy: IfNotPresent ports: - {name: http, containerPort: 8080, protocol: TCP} diff --git a/services/hermes/chat-statefulset.yaml b/services/hermes/chat-statefulset.yaml index 6ad5eff3..dda35cd7 100644 --- a/services/hermes/chat-statefulset.yaml +++ b/services/hermes/chat-statefulset.yaml @@ -29,7 +29,7 @@ spec: ai.bstein.dev/router-wire-contract: ollama-numeric-keepalive ai.bstein.dev/isolation: one Hermes process and PVC per Keycloak subject ai.bstein.dev/model-policy: uniform automatic policy with per-user overrides - ai.bstein.dev/config-rev: "20260815-runtime-access-boundary" + ai.bstein.dev/config-rev: "20260816-telegram-topics" vault.hashicorp.com/agent-inject: "true" vault.hashicorp.com/role: hermes-chat vault.hashicorp.com/agent-inject-secret-chat-relay-key: kv/data/atlas/hermes/chat-telegram @@ -196,6 +196,29 @@ spec: resources: requests: {cpu: 25m, memory: 64Mi} limits: {cpu: 100m, memory: 128Mi} + - name: patch-api-server-sessions + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/bin/sh, -ec] + args: + - | + /opt/hermes/.venv/bin/python /opt/coordinator/patch_api_server_sessions.py \ + /opt/hermes/gateway/platforms/api_server.py /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: + allowPrivilegeEscalation: false + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - {name: home, mountPath: /opt/data} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: api-server-patch, mountPath: /patched} + resources: + requests: {cpu: 25m, memory: 64Mi} + limits: {cpu: 100m, memory: 128Mi} - name: patch-subprocess-secret-boundary image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 imagePullPolicy: IfNotPresent @@ -260,6 +283,7 @@ spec: - {name: workspace, mountPath: /opt/data/workspace} - {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: 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} @@ -285,7 +309,7 @@ spec: requests: {cpu: 250m, memory: 512Mi} limits: {cpu: "1", memory: 2Gi} - name: webui - image: registry.bstein.dev/bstein/hermes-webui@sha256:9c2fe8341c7b650e08d10acead3151b19e2af737863268bafb39b3d9517575b1 + image: registry.bstein.dev/bstein/hermes-webui@sha256:ac6ba7bfd8a86227f31a9a96ebea41227ccf70391f7dcf34206f4d58e835e50e imagePullPolicy: IfNotPresent command: [/bin/sh, -ec] args: @@ -403,6 +427,8 @@ spec: defaultMode: 0555 - name: auth-patch emptyDir: {} + - name: api-server-patch + emptyDir: {} - name: stream-recovery-patch emptyDir: {} - name: subprocess-secret-patch diff --git a/services/hermes/deployment.yaml b/services/hermes/deployment.yaml index 944cf381..25b3ada3 100644 --- a/services/hermes/deployment.yaml +++ b/services/hermes/deployment.yaml @@ -379,7 +379,7 @@ spec: cpu: "2" memory: 4Gi - name: webui - image: registry.bstein.dev/bstein/hermes-webui@sha256:9c2fe8341c7b650e08d10acead3151b19e2af737863268bafb39b3d9517575b1 + image: registry.bstein.dev/bstein/hermes-webui@sha256:ac6ba7bfd8a86227f31a9a96ebea41227ccf70391f7dcf34206f4d58e835e50e imagePullPolicy: IfNotPresent command: [/bin/sh, -ec] args: diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index ae7d93ad..3b9c2345 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -80,6 +80,7 @@ configMapGenerator: - kanban_status_recovery.py=scripts/kanban_status_recovery.py - migrate_herdr_state.py=scripts/migrate_herdr_state.py - migrate_api_session_lineage.py=scripts/migrate_api_session_lineage.py + - migrate_telegram_api_sessions.py=scripts/migrate_telegram_api_sessions.py - patch_api_server_sessions.py=scripts/patch_api_server_sessions.py - patch_web_session_activity.py=scripts/patch_web_session_activity.py - patch_subprocess_secret_boundary.py=scripts/patch_subprocess_secret_boundary.py diff --git a/services/hermes/router/main.go b/services/hermes/router/main.go index 50201aa1..16d7656c 100644 --- a/services/hermes/router/main.go +++ b/services/hermes/router/main.go @@ -26,12 +26,23 @@ type linkRecord struct { ExpiresAt int64 `json:"expires_at"` } +type telegramTopic struct { + Label string `json:"label"` + LastUsed int64 `json:"last_used"` +} + +type telegramTopicState struct { + Active string `json:"active,omitempty"` + Topics map[string]telegramTopic `json:"topics,omitempty"` +} + type tenantState struct { - Salt string `json:"salt"` - Assignments map[string]int `json:"assignments"` - Telegram map[string]int `json:"telegram,omitempty"` - LinkCodes map[string]linkRecord `json:"link_codes,omitempty"` - TelegramOffset int64 `json:"telegram_offset,omitempty"` + Salt string `json:"salt"` + Assignments map[string]int `json:"assignments"` + Telegram map[string]int `json:"telegram,omitempty"` + TelegramTopics map[string]telegramTopicState `json:"telegram_topics,omitempty"` + LinkCodes map[string]linkRecord `json:"link_codes,omitempty"` + TelegramOffset int64 `json:"telegram_offset,omitempty"` } type tenantRouter struct { @@ -76,9 +87,10 @@ func newTenantRouter(statePath string, slots int, backendURL func(int) string) ( backendURL: backendURL, now: time.Now, state: tenantState{ - Assignments: map[string]int{}, - Telegram: map[string]int{}, - LinkCodes: map[string]linkRecord{}, + Assignments: map[string]int{}, + Telegram: map[string]int{}, + TelegramTopics: map[string]telegramTopicState{}, + LinkCodes: map[string]linkRecord{}, }, } content, err := os.ReadFile(statePath) @@ -95,6 +107,9 @@ func newTenantRouter(statePath string, slots int, backendURL func(int) string) ( if router.state.Telegram == nil { router.state.Telegram = map[string]int{} } + if router.state.TelegramTopics == nil { + router.state.TelegramTopics = map[string]telegramTopicState{} + } if router.state.LinkCodes == nil { router.state.LinkCodes = map[string]linkRecord{} } @@ -196,7 +211,13 @@ func (router *tenantRouter) consumeLink(telegramUser, code string) (int, error) return 0, fmt.Errorf("link code is invalid or expired") } delete(router.state.LinkCodes, digest) - router.state.Telegram[router.identityHash("telegram", telegramUser)] = record.Slot + identity := router.identityHash("telegram", telegramUser) + if previousSlot, linked := router.state.Telegram[identity]; linked && previousSlot != record.Slot { + // A link moved to a different tenant must not carry topic labels or the + // active conversation selector across that isolation boundary. + delete(router.state.TelegramTopics, identity) + } + router.state.Telegram[identity] = record.Slot if err := router.saveLocked(); err != nil { return 0, err } @@ -235,6 +256,7 @@ func (router *tenantRouter) unlinkTelegram(subject string) error { for identity, linkedSlot := range router.state.Telegram { if linkedSlot == slot { delete(router.state.Telegram, identity) + delete(router.state.TelegramTopics, identity) } } for digest, record := range router.state.LinkCodes { diff --git a/services/hermes/router/telegram.go b/services/hermes/router/telegram.go index a9dca006..c4cc3b66 100644 --- a/services/hermes/router/telegram.go +++ b/services/hermes/router/telegram.go @@ -276,7 +276,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 /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. 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) @@ -284,6 +284,30 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) { _ = bot.sendText(message.Chat.ID, "This Telegram account is not linked. Sign in to chat.hermes.bstein.dev and open Telegram to connect it.") return } + if command == "topic" { + if len(args) == 0 { + active := bot.router.activeTelegramTopic(userID) + _ = bot.sendText(message.Chat.ID, "Current topic: "+active.Label+". Use /topic to switch or create one.") + return + } + active, err := bot.router.selectTelegramTopic(userID, strings.Join(args, " ")) + if err != nil { + _ = bot.sendText(message.Chat.ID, "That topic name is invalid. Use a short descriptive name, up to 48 characters.") + return + } + _ = bot.sendText(message.Chat.ID, "Switched to topic: "+active.Label+". Messages here keep their own bounded context.") + return + } + if command == "topics" { + active := bot.router.activeTelegramTopic(userID) + lines := []string{"Telegram topics (current: " + active.Label + "):"} + for _, topic := range bot.router.telegramTopics(userID) { + lines = append(lines, "• "+topic.Label) + } + lines = append(lines, "Use /topic to switch or create one.") + _ = bot.sendText(message.Chat.ID, strings.Join(lines, "\n")) + return + } text := strings.TrimSpace(message.Text) hasPhoto := len(message.Photo) > 0 if hasPhoto { @@ -303,9 +327,13 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) { lock := bot.slotLock(slot) lock.Lock() defer lock.Unlock() + topic, err := bot.router.touchTelegramTopic(userID) + if err != nil { + _ = 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") var reply string - var err error if hasPhoto { media, fetchErr := bot.fetchTelegramPhoto(message.Photo) if fetchErr != nil { @@ -313,9 +341,9 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) { return } defer os.Remove(media.Path) - reply, err = bot.askTenantImage(slot, text, media, update.UpdateID) + reply, err = bot.askTenantImage(slot, text, media, update.UpdateID, topic) } else { - reply, err = bot.askTenant(slot, text, update.UpdateID) + reply, err = bot.askTenant(slot, text, update.UpdateID, topic) } if err != nil { _ = bot.sendText(message.Chat.ID, "Hermes could not answer right now. Please try again shortly.") @@ -324,19 +352,20 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) { _ = bot.sendReply(message.Chat.ID, slot, reply) } -func (bot *telegramBot) askTenant(slot int, text string, updateID int64) (string, error) { +func (bot *telegramBot) askTenant(slot int, text string, updateID int64, topic activeTelegramTopic) (string, error) { if bot.router.backendAPIURL == nil { return "", errors.New("tenant API unavailable") } payload, _ := json.Marshal(map[string]any{ "input": text, - "conversation": "telegram", + "conversation": topic.Conversation, "store": true, + "truncation": "auto", }) - return bot.askTenantRequest(slot, bytes.NewReader(payload), updateID) + return bot.askTenantRequest(slot, bytes.NewReader(payload), updateID, topic) } -func (bot *telegramBot) askTenantRequest(slot int, bodyReader io.Reader, updateID int64) (string, error) { +func (bot *telegramBot) askTenantRequest(slot int, bodyReader io.Reader, updateID int64, topic activeTelegramTopic) (string, error) { if bot.router.backendAPIURL == nil { return "", errors.New("tenant API unavailable") } @@ -353,7 +382,9 @@ func (bot *telegramBot) askTenantRequest(slot int, bodyReader io.Reader, updateI } request.Header.Set("Authorization", "Bearer "+bot.config.RelayKey) request.Header.Set("Content-Type", "application/json") - request.Header.Set("X-Hermes-Session-Key", "telegram") + request.Header.Set("X-Hermes-Session-Key", topic.Conversation) + request.Header.Set("X-Hermes-Conversation-Platform", "telegram") + request.Header.Set("X-Hermes-Conversation-Title", "Telegram · "+topic.Label) request.Header.Set("Idempotency-Key", "telegram-"+strconv.FormatInt(updateID, 10)) response, err := bot.agentClient.Do(request) if err != nil { @@ -463,6 +494,8 @@ func (router *tenantRouter) setTelegramOffset(offset int64) error { func (router *tenantRouter) unlinkTelegramUser(userID string) error { router.mu.Lock() defer router.mu.Unlock() - delete(router.state.Telegram, router.identityHash("telegram", userID)) + identity := router.identityHash("telegram", userID) + delete(router.state.Telegram, identity) + delete(router.state.TelegramTopics, identity) return router.saveLocked() } diff --git a/services/hermes/router/telegram_input.go b/services/hermes/router/telegram_input.go index ff9c1f79..40be7f96 100644 --- a/services/hermes/router/telegram_input.go +++ b/services/hermes/router/telegram_input.go @@ -145,7 +145,7 @@ func (bot *telegramBot) fetchTelegramPhoto(photos []telegramPhotoSize) (telegram return telegramInputMedia{MIME: mimeType, Path: temporaryPath}, nil } -func streamTelegramImageRequest(writer io.Writer, prompt string, media telegramInputMedia) error { +func streamTelegramImageRequest(writer io.Writer, prompt string, media telegramInputMedia, topic activeTelegramTopic) error { promptJSON, err := json.Marshal(prompt) if err != nil { return err @@ -170,19 +170,23 @@ func streamTelegramImageRequest(writer io.Writer, prompt string, media telegramI if closeFileErr != nil { return closeFileErr } - _, err = io.WriteString(writer, `"}]}],"conversation":"telegram","store":true}`) + conversationJSON, err := json.Marshal(topic.Conversation) + if err != nil { + return err + } + _, err = fmt.Fprintf(writer, `"}]}],"conversation":%s,"store":true,"truncation":"auto"}`, conversationJSON) return err } -func (bot *telegramBot) askTenantImage(slot int, prompt string, media telegramInputMedia, updateID int64) (string, error) { +func (bot *telegramBot) askTenantImage(slot int, prompt string, media telegramInputMedia, updateID int64, topic activeTelegramTopic) (string, error) { reader, writer := io.Pipe() writeDone := make(chan error, 1) go func() { - err := streamTelegramImageRequest(writer, prompt, media) + err := streamTelegramImageRequest(writer, prompt, media, topic) _ = writer.CloseWithError(err) writeDone <- err }() - answer, requestErr := bot.askTenantRequest(slot, reader, updateID) + answer, requestErr := bot.askTenantRequest(slot, reader, updateID, topic) _ = reader.Close() writeErr := <-writeDone if requestErr != nil { diff --git a/services/hermes/router/telegram_test.go b/services/hermes/router/telegram_test.go index 73b852d6..317dcadc 100644 --- a/services/hermes/router/telegram_test.go +++ b/services/hermes/router/telegram_test.go @@ -35,14 +35,17 @@ func TestAskTenantUsesAuthenticatedNamedConversation(t *testing.T) { if request.Header.Get("Authorization") != "Bearer relay-secret" { t.Fatal("relay authentication was not set") } - if request.Header.Get("X-Hermes-Session-Key") != "telegram" { + if request.Header.Get("X-Hermes-Session-Key") != "telegram-topic-cassandra" { t.Fatal("Telegram session scope was not set") } + if request.Header.Get("X-Hermes-Conversation-Platform") != "telegram" || request.Header.Get("X-Hermes-Conversation-Title") != "Telegram · Cassandra" { + t.Fatalf("Telegram origin metadata was not set: %#v", request.Header) + } var payload map[string]any if err := json.NewDecoder(request.Body).Decode(&payload); err != nil { t.Fatal(err) } - if payload["conversation"] != "telegram" || payload["input"] != "hello" { + if payload["conversation"] != "telegram-topic-cassandra" || payload["input"] != "hello" || payload["truncation"] != "auto" { t.Fatalf("unexpected payload: %#v", payload) } writer.Header().Set("Content-Type", "application/json") @@ -55,7 +58,10 @@ func TestAskTenantUsesAuthenticatedNamedConversation(t *testing.T) { } router.backendAPIURL = func(slot int) string { return server.URL } bot := newTelegramBot(telegramConfig{RelayKey: "relay-secret"}, router) - answer, err := bot.askTenant(0, "hello", 42) + answer, err := bot.askTenant(0, "hello", 42, activeTelegramTopic{ + Conversation: "telegram-topic-cassandra", + Label: "Cassandra", + }) if err != nil { t.Fatal(err) } @@ -183,11 +189,12 @@ func TestTelegramImageRequestStreamsMultimodalInput(t *testing.T) { } `json:"input"` Conversation string `json:"conversation"` Store bool `json:"store"` + Truncation string `json:"truncation"` } if err := json.Unmarshal(body, &payload); err != nil { t.Fatalf("decode streamed request: %v\n%s", err, body) } - if payload.Conversation != "telegram" || !payload.Store || len(payload.Input) != 1 || len(payload.Input[0].Content) != 2 { + if payload.Conversation != "telegram-topic-images" || !payload.Store || payload.Truncation != "auto" || len(payload.Input) != 1 || len(payload.Input[0].Content) != 2 { t.Fatalf("unexpected image payload: %#v", payload) } if payload.Input[0].Role != "user" || payload.Input[0].Content[0].Text != "What is this?" || payload.Input[0].Content[1].Type != "input_image" { @@ -212,7 +219,7 @@ func TestTelegramImageRequestStreamsMultimodalInput(t *testing.T) { answer, err := bot.askTenantImage(0, "What is this?", telegramInputMedia{ MIME: "image/jpeg", Path: imagePath, - }, 99) + }, 99, activeTelegramTopic{Conversation: "telegram-topic-images", Label: "Images"}) if err != nil { t.Fatal(err) } @@ -221,6 +228,51 @@ func TestTelegramImageRequestStreamsMultimodalInput(t *testing.T) { } } +func TestTelegramTopicsPersistWithoutStoringTelegramIdentity(t *testing.T) { + statePath := filepath.Join(t.TempDir(), "state.json") + router, err := newTenantRouter(statePath, 1, func(int) string { return "" }) + if err != nil { + t.Fatal(err) + } + selected, err := router.selectTelegramTopic("123456789", " Cassandra repairs ") + if err != nil { + t.Fatal(err) + } + if selected.Label != "Cassandra repairs" || !strings.HasPrefix(selected.Conversation, "telegram-topic-") { + t.Fatalf("unexpected selected topic: %#v", selected) + } + reloaded, err := newTenantRouter(statePath, 1, func(int) string { return "" }) + if err != nil { + t.Fatal(err) + } + if active := reloaded.activeTelegramTopic("123456789"); active != selected { + t.Fatalf("topic did not persist: got %#v want %#v", active, selected) + } + content, err := os.ReadFile(statePath) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(content), "123456789") { + t.Fatal("raw Telegram identity leaked into router state") + } +} + +func TestTelegramTopicValidationAndLegacyGeneralConversation(t *testing.T) { + if telegramTopicConversation("general") != "telegram" { + t.Fatal("general topic did not preserve the legacy conversation") + } + if _, err := normalizeTelegramTopicLabel(strings.Repeat("x", 49)); err == nil { + t.Fatal("oversized topic name was accepted") + } + if _, err := normalizeTelegramTopicLabel("unsafe\nname"); err != nil { + // Whitespace is deliberately normalized so a pasted line break is safe. + t.Fatalf("normalizable whitespace was rejected: %v", err) + } + if first, second := telegramTopicID("Cassandra"), telegramTopicID("cassandra"); first != second { + t.Fatalf("case-insensitive topic IDs diverged: %q %q", first, second) + } +} + 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 new file mode 100644 index 00000000..d5e9bed6 --- /dev/null +++ b/services/hermes/router/telegram_topics.go @@ -0,0 +1,139 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "sort" + "strings" + "unicode" + "unicode/utf8" +) + +const telegramDefaultTopic = "General" + +type activeTelegramTopic struct { + Conversation string + Label string +} + +func normalizeTelegramTopicLabel(raw string) (string, error) { + label := strings.Join(strings.Fields(strings.TrimSpace(raw)), " ") + if label == "" { + return "", errors.New("topic name is empty") + } + if utf8.RuneCountInString(label) > 48 { + return "", errors.New("topic name is too long") + } + for _, character := range label { + if unicode.IsControl(character) { + return "", errors.New("topic name contains control characters") + } + } + return label, nil +} + +func telegramTopicID(label string) string { + if strings.EqualFold(strings.TrimSpace(label), telegramDefaultTopic) { + return "general" + } + digest := sha256.Sum256([]byte(strings.ToLower(strings.TrimSpace(label)))) + return hex.EncodeToString(digest[:8]) +} + +func telegramTopicConversation(topicID string) string { + if topicID == "" || topicID == "general" { + // Preserve the original named conversation so existing Telegram context + // survives rollout into topic-aware routing. + return "telegram" + } + return "telegram-topic-" + topicID +} + +func (router *tenantRouter) activeTelegramTopic(userID string) activeTelegramTopic { + identity := router.identityHash("telegram", userID) + router.mu.Lock() + defer router.mu.Unlock() + state := router.state.TelegramTopics[identity] + topicID := state.Active + if topicID == "" { + topicID = "general" + } + label := telegramDefaultTopic + if topic, found := state.Topics[topicID]; found && strings.TrimSpace(topic.Label) != "" { + label = topic.Label + } + return activeTelegramTopic{Conversation: telegramTopicConversation(topicID), Label: label} +} + +func (router *tenantRouter) selectTelegramTopic(userID, rawLabel string) (activeTelegramTopic, error) { + label, err := normalizeTelegramTopicLabel(rawLabel) + if err != nil { + return activeTelegramTopic{}, err + } + identity := router.identityHash("telegram", userID) + topicID := telegramTopicID(label) + if topicID == "general" { + label = telegramDefaultTopic + } + router.mu.Lock() + defer router.mu.Unlock() + state := router.state.TelegramTopics[identity] + if state.Topics == nil { + state.Topics = map[string]telegramTopic{} + } + state.Active = topicID + state.Topics[topicID] = telegramTopic{Label: label, LastUsed: router.now().Unix()} + router.state.TelegramTopics[identity] = state + if err := router.saveLocked(); err != nil { + return activeTelegramTopic{}, err + } + return activeTelegramTopic{Conversation: telegramTopicConversation(topicID), Label: label}, nil +} + +func (router *tenantRouter) touchTelegramTopic(userID string) (activeTelegramTopic, error) { + identity := router.identityHash("telegram", userID) + router.mu.Lock() + defer router.mu.Unlock() + state := router.state.TelegramTopics[identity] + if state.Topics == nil { + state.Topics = map[string]telegramTopic{} + } + topicID := state.Active + if topicID == "" { + topicID = "general" + state.Active = topicID + } + topic := state.Topics[topicID] + if strings.TrimSpace(topic.Label) == "" { + topic.Label = telegramDefaultTopic + } + topic.LastUsed = 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 +} + +func (router *tenantRouter) telegramTopics(userID string) []telegramTopic { + identity := router.identityHash("telegram", userID) + router.mu.Lock() + defer router.mu.Unlock() + state := router.state.TelegramTopics[identity] + topics := make([]telegramTopic, 0, len(state.Topics)+1) + for _, topic := range state.Topics { + topics = append(topics, topic) + } + if _, found := state.Topics["general"]; !found { + topics = append(topics, telegramTopic{Label: telegramDefaultTopic}) + } + sort.SliceStable(topics, func(left, right int) bool { + if topics[left].LastUsed == topics[right].LastUsed { + return strings.ToLower(topics[left].Label) < strings.ToLower(topics[right].Label) + } + return topics[left].LastUsed > topics[right].LastUsed + }) + return topics +} diff --git a/services/hermes/scripts/migrate_telegram_api_sessions.py b/services/hermes/scripts/migrate_telegram_api_sessions.py new file mode 100644 index 00000000..82f58456 --- /dev/null +++ b/services/hermes/scripts/migrate_telegram_api_sessions.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Backfill Telegram origin metadata for durable API conversations.""" + +from __future__ import annotations + +import argparse +import json +import sqlite3 +from pathlib import Path + + +def telegram_sessions(response_store: Path) -> dict[str, str]: + """Return session ID to stable Telegram conversation-key mappings.""" + if not response_store.exists(): + return {} + connection = sqlite3.connect(f"file:{response_store}?mode=ro", uri=True) + try: + rows = connection.execute( + """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-%'""" + ).fetchall() + 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): + continue + if session_id: + result[session_id] = str(conversation) + return result + + +def migrate(state_database: Path, response_store: Path) -> int: + """Annotate known sessions without creating or rewriting conversations.""" + mappings = telegram_sessions(response_store) + if not mappings or not state_database.exists(): + return 0 + connection = sqlite3.connect(state_database) + changed = 0 + try: + columns = { + str(row[1]) + for row in connection.execute("PRAGMA table_info(sessions)").fetchall() + } + required = { + "id", "source", "session_key", "chat_type", "display_name", + "origin_json", "title", + } + if not required.issubset(columns): + return 0 + for session_id, conversation in mappings.items(): + origin = json.dumps( + {"platform": "telegram", "session_key": conversation}, + separators=(",", ":"), + ) + 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, ?) + 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), + ) + changed += max(cursor.rowcount, 0) + connection.commit() + finally: + connection.close() + return changed + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("state_database", type=Path) + parser.add_argument("response_store", type=Path) + args = parser.parse_args() + migrate(args.state_database, args.response_store) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/hermes/scripts/patch_api_server_sessions.py b/services/hermes/scripts/patch_api_server_sessions.py index 44e3d52e..bc7cafcc 100644 --- a/services/hermes/scripts/patch_api_server_sessions.py +++ b/services/hermes/scripts/patch_api_server_sessions.py @@ -53,6 +53,72 @@ RUNS_BEFORE = ''' run_id = f"run_{uuid.uuid4().hex}" # Approval queues gate host-side tool execution and must be isolated ''' +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 + # groups the entire conversation under one session entry. + session_id = stored_session_id or str(uuid.uuid4()) + + # A trusted relay may identify a conversational surface so state.db and + # 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() + conversation_title = request.headers.get( + "X-Hermes-Conversation-Title", "" + ).strip() + if conversation_platform: + if conversation_platform != "telegram": + return web.json_response( + _openai_error("Unsupported conversation platform"), status=400 + ) + if ( + not gateway_session_key + or not gateway_session_key.startswith("telegram") + or len(conversation_title) > 128 + or re.search(r'[\\r\\n\\x00]', conversation_title) + ): + return web.json_response( + _openai_error("Invalid conversation metadata"), status=400 + ) + db = self._ensure_session_db() + if db is None: + return web.json_response( + _openai_error("Session database unavailable"), status=503 + ) + origin = json.dumps({ + "platform": "telegram", + "session_key": gateway_session_key, + }, separators=(",", ":")) + db.create_session( + session_id, + "api_server", + model=str(body.get("model") or self._model_name or ""), + system_prompt=instructions if isinstance(instructions, str) else None, + session_key=gateway_session_key, + chat_type="private", + ) + db.record_gateway_session_peer( + session_id, + source="api_server", + session_key=gateway_session_key, + chat_type="private", + display_name="Telegram", + origin_json=origin, + ) + if conversation_title: + db.set_session_title(session_id, conversation_title) + + # Per-client model routing for /v1/responses (see model_routes). +''' + RUNS_AFTER = ''' run_id = f"run_{uuid.uuid4().hex}" session_id = body.get("session_id") or stored_session_id or run_id @@ -362,6 +428,8 @@ def patch(source: Path, destination: Path) -> None: raise RuntimeError("Hermes API runs patch context changed") if RUN_CLOSE_BEFORE not in content: raise RuntimeError("Hermes API run-close patch context changed") + if RESPONSES_SESSION_BEFORE not in content: + raise RuntimeError("Hermes Responses session metadata patch context changed") for marker, message in ( (EVENT_CALLBACK_SIGNATURE_BEFORE, "event callback signature"), (EVENT_CALLBACK_BODY_BEFORE, "event callback body"), @@ -375,6 +443,7 @@ def patch(source: Path, destination: Path) -> None: content = content.replace(BEFORE, AFTER, 1) 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( EVENT_CALLBACK_SIGNATURE_BEFORE, EVENT_CALLBACK_SIGNATURE_AFTER, diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index f4a9a64f..c286ac38 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -318,6 +318,12 @@ def test_webui_recovers_auth_and_labels_session_scoped_controls(): assert "window.location.assign('/oauth2/start?rd='" in dockerfile assert "childrenExpanded?'▾ ':'▸ '" in dockerfile assert "'atlas/auto/maximum': 'Automatic · Maximum'" in dockerfile + assert "hermes-webui-telegram-project-patch.py" in dockerfile + telegram_project_patch = ( + ROOT / "dockerfiles" / "hermes-webui-telegram-project-patch.py" + ).read_text() + assert "TELEGRAM_PROJECT_NAME = 'Telegram'" in telegram_project_patch + assert "session_key.startswith('telegram-topic-')" in telegram_project_patch router = (ROOT / "dockerfiles" / "hermes-webui-router.js").read_text() assert "'atlas/auto/fast':'AUTO · Fast'" in router @@ -676,7 +682,7 @@ def test_chat_reasoning_uses_switchyard_without_owner_credentials(): statefulset = _documents(HERMES / "chat-statefulset.yaml")[0] assert statefulset["spec"]["template"]["metadata"]["annotations"][ "ai.bstein.dev/config-rev" - ] == "20260815-runtime-access-boundary" + ] == "20260816-telegram-topics" pod_spec = statefulset["spec"]["template"]["spec"] patch_init = next( item for item in pod_spec["initContainers"] @@ -696,6 +702,20 @@ def test_chat_reasoning_uses_switchyard_without_owner_credentials(): "mountPath": "/opt/hermes/agent/conversation_loop.py", "subPath": "conversation_loop.py", } in hermes["volumeMounts"] + api_session_init = next( + item for item in pod_spec["initContainers"] + if item["name"] == "patch-api-server-sessions" + ) + assert "patch_api_server_sessions.py" in api_session_init["args"][0] + assert "migrate_telegram_api_sessions.py" in api_session_init["args"][0] + assert { + "name": "api-server-patch", + "mountPath": "/opt/hermes/gateway/platforms/api_server.py", + "subPath": "api_server.py", + } in hermes["volumeMounts"] + assert any( + volume["name"] == "api-server-patch" for volume in pod_spec["volumes"] + ) assert not any( mount["mountPath"].endswith("/.codex") for mount in hermes["volumeMounts"] @@ -1231,6 +1251,7 @@ def test_api_session_patch_accepts_parent_lineage(tmp_path: Path): + module.RUNS_BEFORE + "run body\n" + module.RUN_CLOSE_BEFORE + + module.RESPONSES_SESSION_BEFORE + module.EVENT_CALLBACK_SIGNATURE_BEFORE + "callback docstring and push helper\n" + module.EVENT_CALLBACK_BODY_BEFORE @@ -1251,6 +1272,11 @@ def test_api_session_patch_accepts_parent_lineage(tmp_path: Path): assert "HERMES_API_DEFAULT_PARENT_MATCH_PREFIXES" in patched assert "user_message.startswith(default_prefixes)" in patched assert "session_parent_conflict" in patched + assert "X-Hermes-Conversation-Platform" in patched + assert "X-Hermes-Conversation-Title" in patched + assert 'conversation_platform != "telegram"' in patched + assert "db.record_gateway_session_peer(" in patched + assert 'display_name="Telegram"' in patched assert "db.reopen_session(session_id)" in patched assert 'db.end_session(session_id, f"api_run_{terminal_status}")' in patched assert "def _record_run_activity(" in patched @@ -1276,6 +1302,64 @@ def test_api_session_patch_accepts_parent_lineage(tmp_path: Path): ) +def test_telegram_api_session_migration_is_bounded_and_idempotent(tmp_path: Path): + """Only named Telegram conversations receive presentation metadata.""" + module_path = HERMES / "scripts" / "migrate_telegram_api_sessions.py" + spec = importlib.util.spec_from_file_location("migrate_telegram_sessions", module_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + 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 (id, source) VALUES (?, ?)", + (("telegram-session", "api_server"), ("other-session", "api_server")), + ) + 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)", + ( + ("telegram-response", json.dumps({"session_id": "telegram-session"})), + ("other-response", json.dumps({"session_id": "other-session"})), + ), + ) + connection.executemany( + "INSERT INTO conversations VALUES (?, ?)", + (("telegram", "telegram-response"), ("unrelated", "other-response")), + ) + + assert module.migrate(state, responses) == 1 + assert module.migrate(state, responses) == 0 + with sqlite3.connect(state) as connection: + telegram = connection.execute( + "SELECT session_key, chat_type, display_name, origin_json, title " + "FROM sessions WHERE id = 'telegram-session'" + ).fetchone() + other = connection.execute( + "SELECT session_key, title FROM sessions WHERE id = 'other-session'" + ).fetchone() + assert telegram[:3] == ("telegram", "private", "Telegram") + assert json.loads(telegram[3]) == { + "platform": "telegram", + "session_key": "telegram", + } + assert telegram[4] == "Telegram · General" + assert other == (None, None) + + def test_web_session_activity_patch_projects_bounded_events(tmp_path: Path): """The DOM transcript includes events without polluting agent history.""" module_path = HERMES / "scripts" / "patch_web_session_activity.py"