hermes: add durable Telegram topics

This commit is contained in:
jenkins 2026-08-16 15:59:09 -03:00
parent 9eee5cb339
commit 0dd6ea0f02
14 changed files with 673 additions and 34 deletions

View File

@ -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 \

View File

@ -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),",
)

View File

@ -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}

View File

@ -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

View File

@ -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:

View File

@ -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

View File

@ -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 {

View File

@ -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 <name> to create or switch a durable topic, /topics to list topics, or /unlink to disconnect this Telegram account. Model and intensity controls are available in Hermes WebUI.")
return
}
slot, linked := bot.router.telegramSlot(userID)
@ -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 <name> 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 <name> 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()
}

View File

@ -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 {

View File

@ -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 {

View File

@ -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
}

View File

@ -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())

View File

@ -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,

View File

@ -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"