This repository has no description
37 kB
841 lines
1"""hermes-voice: talk to Hermes through OpenAI's full-duplex gpt-live-1.
2
3The browser holds a WebRTC session with gpt-live-1, which listens and speaks at
4the same time and chats on its own. Whenever the user asks for real work, the
5voice model emits a client delegation; the page hands the recent transcript to
6this server, which starts it as a Hermes run on the API server (POST /v1/runs,
7one Hermes session id per conversation), follows the run's event stream and
8streams the reply back. The page then gives the answer to the voice model,
9which says it aloud.
10
11A run lives independently of any HTTP connection, and this server follows
12every run it started to its end, whether or not a page is still listening. A
13newer delegation within the same call stops the previous run (the user
14corrected or replaced the request); hanging up just ends the call, and the run
15finishes whatever the user asked for. On hang-up the page sends what the user
16said since the last delegation, so a request spoken right before hanging up
17still reaches Hermes. When a run asks for approval while no page is listening,
18the question goes to the owner's Matrix DM (the gateway's Matrix home room,
19through its webhook platform), and the hermes-voice plugin's
20`/assist approve|deny <code>` command answers it.
21
22This server exists so neither secret reaches the browser: it exchanges the
23page's SDP offer for a Live session with the OpenAI key, and it calls Hermes
24with the API server's bearer token. That token is root-equivalent for the
25Hermes host, so whoever can use this server can drive the agent: every API
26route requires the page's own access token (--access-token-file), and serving
27beyond loopback without one is refused.
28
29Vendor contract: https://developers.openai.com/api/docs/guides/live-delegation
30"""
31
32import argparse
33import hashlib
34import hmac
35import json
36import logging
37import os
38import re
39import secrets
40import socket
41import threading
42import time
43import urllib.error
44import urllib.request
45from http import HTTPStatus
46from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
47from pathlib import Path
48
49log = logging.getLogger("hermes-voice")
50
51OPENAI_BASE_URL = "https://api.openai.com/v1"
52DEFAULT_MODEL = "gpt-live-1"
53DEFAULT_VOICE = "marin"
54# gpt-live-1 voices as OpenAI's voice picker lists them: language, regional
55# influence, presentation and character, natural voices first, then the
56# generated ones. The docs' voice table and the openai SDK (3.28.0) list only
57# part of them; /v1/live/sessions accepts the German, French, Hindi, Japanese
58# and Korean voices as well, while an unknown name is refused with 403.
59BUILT_IN_VOICES = {
60 "gleam": "English, North American, female: warm and conversational",
61 "ripple": "English, Australian, male: warm and conversational",
62 "vesper": "English, British, male: clear and direct",
63 "willow": "English, Irish, female: expressive and bright",
64 "stone": "English, Irish, male: calm and precise",
65 "meridian": "English, North American, male: grounded and measured",
66 "bossa": "Portuguese, Brazilian, female: clear and direct",
67 "tempo": "Portuguese, Brazilian, male: grounded and measured",
68 "cedar": "English, North American, male: open and upbeat",
69 "marin": "English, North American, female: animated and earnest",
70 "nuri": "German, female: warm and animated",
71 "juni": "German, female: bright and expressive",
72 "flitz": "German, male: rich and playful",
73 "brise": "French, female: clear and conversational",
74 "sillage": "French, male: warm and grounded",
75 "nira": "Hindi, female: soft and thoughtful",
76 "harema": "Japanese, female: gentle and engaging",
77 "shida": "Japanese, male: quiet and grounded",
78 "noeul": "Korean, female: airy and expressive",
79 "quartz": "English, Australian, female, generated: warm and conversational",
80 "alloy": "English, North American, male, generated: balanced and clear",
81 "ash": "English, North American, male, generated: savvy and relaxed",
82 "ballad": "English, North American, male, generated: cheerful and candid",
83 "coral": "English, North American, female, generated: easygoing and versatile",
84 "echo": "English, North American, male, generated: bright and inquisitive",
85 "sage": "English, North American, female, generated: confident and optimistic",
86 "shimmer": "English, North American, female, generated: composed and direct",
87 "verse": "English, North American, male, generated: calm and affirming",
88 "beacon": "English, Filipino, male, generated: clear and personable",
89 "delta": "English, Southern U.S., female, generated: warm and easygoing",
90 "cinder": "English, Southern U.S., male, generated: steady and grounded",
91}
92# gpt-live-1 bills every second a session is open, silence included; the page
93# closes the session after this much inactivity.
94DEFAULT_IDLE_SECONDS = 90
95MAX_BODY = 256 * 1024
96# The run event stream sends a keepalive every 10s, so a read that stalls this
97# long means the API server is gone.
98EVENTS_READ_TIMEOUT = 60
99# Same cadence towards the page: a write is the only way to notice that a page
100# went away while a run is quiet (e.g. waiting for an approval).
101PAGE_KEEPALIVE_SECONDS = 10
102SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
103
104# Frontend persona, adapted from Hermes's own (tools/voice_live.py): role,
105# style and a labelled delegation policy. The backend carries the real
106# instructions, tools and memory.
107PERSONA = """You are Hermes, the user's warm, upbeat and genuinely helpful voice assistant. Sound friendly and glad to help: positive and encouraging, never curt, annoyed, bored or sarcastic, even when delivering bad news or when the user is short with you. Talk fast: speak at a rapid, energetic pace, like someone who talks quickly by nature, with no slow drawn-out words and no long pauses between sentences. Keep replies short; a brief warm acknowledgment before the facts is good.
108
109Backchannel policy: Use moderate backchannels. Acknowledge naturally without competing with the main response.
110
111Interruption policy: Stop speaking when the user interrupts. Listen to what they say.
112
113Delegation policy:
114Backend tools:
115- Hermes agent: a full AI agent with tools. It can run commands on the user's machines, read and edit files, manage the calendar, write and read messages in the user's Matrix chat, browse the web, search, and remember things across sessions. It is the one who actually does work and knows facts.
116
117Delegate to the backend when:
118- The user asks a question that needs facts, current information, or careful reasoning.
119- The user asks you to do, check, find, make, fix, run or remember anything.
120- A correction changes work already requested.
121
122Do not delegate to the backend when:
123- The user greets you, makes small talk, or asks you to repeat a result already provided.
124- You need a brief clarification to understand the request.
125
126Delegate before giving an answer that depends on backend work. Do not guess the result while waiting; say briefly and cheerfully that you are on it, then wait for the result.
127
128Approval policy:
129Hermes sometimes needs the user's approval before it runs a command. When you are told so, explain in plain words what Hermes wants to do and ask whether to approve or deny it. When the user answers, delegate to the backend so the answer reaches Hermes. Never decide an approval yourself."""
130
131# Ephemeral system prompt for every delegated Hermes turn (Hermes layers system
132# messages on top of its core prompt without storing them).
133BACKEND_NOTE = """This turn is a delegation from a live spoken conversation. The user message is a speech transcript: it may contain mis-hearings, phonetic spellings of technical terms, hesitations and later corrections; use the latest intent. Your reply will be spoken aloud by a friendly voice model that paraphrases it: answer in warm, plain conversational sentences, keep it short (a few sentences unless the user asked for detail), no markdown, no lists, no code blocks, no URLs read out character by character. Do the work with your tools as usual; only the final facts need to be spoken. Report an action as done only after it actually succeeded. The user may hang up before your reply is spoken: when they asked for a result or a question somewhere else, such as their Matrix chat (matrix_dm_send), send it there yourself."""
134
135# Instead of BACKEND_NOTE for what the user said right before hanging up, when
136# the voice model had not delegated it yet.
137HUNG_UP_NOTE = """This turn comes from a live spoken conversation the user has already hung up. The user message is a speech transcript of what they said since the last request: it may contain mis-hearings, phonetic spellings of technical terms, hesitations and later corrections; use the latest intent. Nobody hears your reply. If the transcript asks for something, do it with your tools; when the user asked for a result somewhere, such as their Matrix chat (matrix_dm_send), send it there, and if you need something from them, ask in their Matrix chat. If it only says goodbye or thanks, or repeats a request already handled in this conversation, do nothing and reply with a short note saying so."""
138
139
140def glossary_text(glossary):
141 """Glossary entries as prompt lines: `term` or `term (sounds like "…")`."""
142 lines = []
143 for entry in glossary:
144 if isinstance(entry, str):
145 lines.append(f"- {entry}")
146 else:
147 term = entry["term"]
148 spoken = entry.get("spoken")
149 lines.append(f'- {term} (sounds like "{spoken}")' if spoken else f"- {term}")
150 return "\n".join(lines)
151
152
153class Config:
154 def __init__(self, raw, openai_key, hermes_url, hermes_key, matrix_webhook=None):
155 self.matrix_webhook = matrix_webhook
156 self.model = raw.get("model", DEFAULT_MODEL)
157 self.voice = raw.get("voice", DEFAULT_VOICE)
158 self.idle_seconds = int(raw.get("idle_seconds", DEFAULT_IDLE_SECONDS))
159 self.glossary = raw.get("glossary", [])
160 self.extra_instructions = raw.get("instructions", "").strip()
161 self.openai_key = openai_key
162 self.hermes_url = hermes_url.rstrip("/")
163 self.hermes_key = hermes_key
164
165 def live_instructions(self):
166 parts = [PERSONA]
167 if self.glossary:
168 parts.append(
169 "Technical terms the user says often. Understand them when spoken and "
170 "say them like this:\n" + glossary_text(self.glossary)
171 )
172 if self.extra_instructions:
173 parts.append(self.extra_instructions)
174 return "\n\n".join(parts)
175
176 def backend_note(self, hung_up=False):
177 note = HUNG_UP_NOTE if hung_up else BACKEND_NOTE
178 if not self.glossary:
179 return note
180 return (
181 note
182 + "\n\nTerms the transcript may contain as phonetic misspellings:\n"
183 + glossary_text(self.glossary)
184 )
185
186 def voices(self):
187 """Voices the page may pick, name -> description: the built-ins plus
188 the configured one (which may be a custom voice id)."""
189 voices = dict(BUILT_IN_VOICES)
190 voices.setdefault(self.voice, "custom voice")
191 return voices
192
193
194def voice_param(voice):
195 # Custom voices are objects ({"id": "voice_…"}); built-ins are names.
196 if voice.startswith("voice_"):
197 return {"id": voice}
198 return voice
199
200
201def live_history(lines):
202 """Seed history for a new Live session from the page's transcript lines
203 (spoken dialogue only; Hermes answers were spoken through the voice)."""
204 items = []
205 dialogue = [line for line in lines if line.get("speaker") in ("user", "assistant")]
206 for line in dialogue[-24:]:
207 text = str(line.get("text", "")).strip()[:1200]
208 if not text:
209 continue
210 if line["speaker"] == "assistant":
211 content = {"type": "output_text", "text": text}
212 else:
213 content = {"type": "input_text", "text": text}
214 items.append({"type": "message", "role": line["speaker"], "content": [content]})
215 return items
216
217
218def create_live_session(config, sdp, history, voice):
219 session = {
220 "model": config.model,
221 "instructions": config.live_instructions(),
222 "audio": {"output": {"voice": voice_param(voice)}},
223 "delegation": {"type": "client"},
224 # The page is an untrusted frontend: it may only answer delegations and
225 # hang up, not rewrite the session.
226 "client": {
227 "data_channel": {
228 "allowed_client_events": [
229 "session.commentary.append",
230 "session.thinking.append",
231 "session.close",
232 ]
233 }
234 },
235 }
236 if history:
237 session["input"] = history
238 body = json.dumps(
239 {"session": session, "transport": {"type": "webrtc", "sdp": sdp}}
240 ).encode()
241 request = urllib.request.Request(
242 f"{OPENAI_BASE_URL}/live/sessions",
243 data=body,
244 method="POST",
245 headers={
246 "Authorization": f"Bearer {config.openai_key}",
247 "Content-Type": "application/json",
248 },
249 )
250 with urllib.request.urlopen(request, timeout=30) as response:
251 answer = json.load(response)
252 return {"session_id": answer["session"]["id"], "sdp": answer["transport"]["sdp"]}
253
254
255def transcript_message(lines):
256 rendered = []
257 for line in lines:
258 text = str(line.get("text", "")).strip()
259 if text:
260 who = "Assistant" if line.get("speaker") == "assistant" else "User"
261 rendered.append(f"{who}: {text}")
262 return "Spoken conversation since the last request, newest last:\n" + "\n".join(
263 rendered
264 )
265
266
267def iter_sse(response):
268 """Yield (event, data) pairs from a text/event-stream response."""
269 event, data = None, []
270 for raw in response:
271 line = raw.decode("utf-8", "replace").rstrip("\r\n")
272 if not line:
273 if data:
274 yield event, "\n".join(data)
275 event, data = None, []
276 elif line.startswith("event:"):
277 event = line[6:].strip()
278 elif line.startswith("data:"):
279 data.append(line[5:].lstrip())
280 if data:
281 yield event, "\n".join(data)
282
283
284def hermes_request(config, method, path, payload=None, timeout=30):
285 """Open a request to the Hermes API server (the caller closes it)."""
286 request = urllib.request.Request(
287 f"{config.hermes_url}{path}",
288 data=None if payload is None else json.dumps(payload).encode(),
289 method=method,
290 headers={
291 "Authorization": f"Bearer {config.hermes_key}",
292 "Content-Type": "application/json",
293 },
294 )
295 return urllib.request.urlopen(request, timeout=timeout)
296
297
298def hermes_call(config, method, path, payload=None):
299 with hermes_request(config, method, path, payload) as response:
300 return json.load(response)
301
302
303def hermes_start_run(config, hermes_session, lines, hung_up=False):
304 """Start one Hermes turn as a run; return its run id. Hermes queues it
305 behind a run of the same session that is still going."""
306 # A plain string input: Hermes then loads the history of session_id itself.
307 return hermes_call(
308 config,
309 "POST",
310 "/v1/runs",
311 {
312 "input": transcript_message(lines),
313 "instructions": config.backend_note(hung_up),
314 "session_id": hermes_session,
315 },
316 )["run_id"]
317
318
319def hermes_follow_run(config, run_id):
320 """Follow a run's events; yield NDJSON-ready dicts (tool, delta, approval,
321 done, error). Leaving early only detaches: the run keeps going."""
322 with hermes_request(
323 config, "GET", f"/v1/runs/{run_id}/events", timeout=EVENTS_READ_TIMEOUT
324 ) as response:
325 for _, data in iter_sse(response):
326 payload = json.loads(data)
327 event = payload.get("event", "")
328 if event in ("tool.started", "tool.completed"):
329 status = "running" if event == "tool.started" else "completed"
330 if payload.get("error"):
331 status = "error"
332 yield {"type": "tool", "status": status, "label": payload.get("tool")}
333 elif event == "approval.request":
334 # The run blocks until POST /v1/runs/{run_id}/approval answers.
335 # Under the gateway's session authority the request is named
336 # prompt_id and the answer must carry its execution_generation.
337 yield {
338 "type": "approval",
339 "request_id": payload.get("request_id") or payload.get("prompt_id"),
340 "execution_generation": payload.get("execution_generation"),
341 "command": payload.get("command"),
342 "description": payload.get("description"),
343 "timeout_s": payload.get("timeout_s"),
344 }
345 elif event == "run.completed":
346 # Only the final answer: message.delta also streams commentary.
347 yield {"type": "delta", "text": payload.get("output") or ""}
348 yield {"type": "done", "finish_reason": "stop"}
349 return
350 elif event in ("run.failed", "run.cancelled", "run.interrupted"):
351 status = event.removeprefix("run.")
352 error = payload.get("error") or f"Hermes run {status}"
353 yield {"type": "done", "finish_reason": status, "error": error}
354 return
355 yield {"type": "error", "message": "Hermes event stream ended early"}
356
357
358def hermes_approve(config, run_id, approval, choice):
359 """Answer a pending approval of a Hermes run."""
360 answer = {"choice": choice}
361 if approval.get("request_id"):
362 answer["request_id"] = approval["request_id"]
363 if approval.get("execution_generation") is not None:
364 answer["execution_generation"] = approval["execution_generation"]
365 return hermes_call(config, "POST", f"/v1/runs/{run_id}/approval", answer)
366
367
368def notify_matrix(config, text):
369 """Post text to the owner's Matrix DM (the gateway's Matrix home room)
370 through the gateway's webhook platform (a deliver_only route, signed with
371 the API server key)."""
372 body = json.dumps({"text": text}).encode()
373 stamp = str(int(time.time()))
374 signature = hmac.new(
375 config.hermes_key.encode(), stamp.encode() + b"." + body, hashlib.sha256
376 ).hexdigest()
377 request = urllib.request.Request(
378 config.matrix_webhook,
379 data=body,
380 method="POST",
381 headers={
382 "Content-Type": "application/json",
383 "X-Webhook-Timestamp": stamp,
384 "X-Webhook-Signature-V2": signature,
385 },
386 )
387 with urllib.request.urlopen(request, timeout=30):
388 pass
389
390
391# Live runs by run id, and the latest run per Hermes session, which a
392# correction in the same call replaces.
393RUNS = {}
394LATEST = {}
395RUNS_LOCK = threading.Lock()
396# Pending approvals by short code: the page and the Matrix command answer
397# with the code.
398APPROVALS = {}
399APPROVALS_LOCK = threading.Lock()
400
401
402class Run:
403 """A Hermes run this server follows to its end, page or no page. Pages
404 replay its items; an approval nobody can hear goes to Matrix."""
405
406 def __init__(self, config, hermes_session, run_id, hung_up=False):
407 self.config = config
408 self.hermes_session = hermes_session
409 self.run_id = run_id
410 self.items = []
411 self.finished = False
412 self.pages = 0
413 self.hung_up = hung_up
414 self.cond = threading.Condition()
415 with RUNS_LOCK:
416 RUNS[run_id] = self
417 if not hung_up:
418 LATEST[hermes_session] = self
419 threading.Thread(target=self.follow, daemon=True).start()
420
421 def follow(self):
422 started = time.monotonic()
423 tools = 0
424 try:
425 for item in hermes_follow_run(self.config, self.run_id):
426 if item["type"] == "tool" and item["status"] == "running":
427 tools += 1
428 self.add(item)
429 except urllib.error.HTTPError as error:
430 detail = error.read().decode("utf-8", "replace")[:400]
431 self.add({"type": "error", "message": f"Hermes {error.code}: {detail}"})
432 except (urllib.error.URLError, OSError, ValueError) as error:
433 self.add({"type": "error", "message": f"Hermes unreachable: {error}"})
434 settle_approvals(self)
435 with self.cond:
436 self.finished = True
437 self.cond.notify_all()
438 with RUNS_LOCK:
439 RUNS.pop(self.run_id, None)
440 if LATEST.get(self.hermes_session) is self:
441 del LATEST[self.hermes_session]
442 last = self.items[-1] if self.items else {}
443 log.info(
444 "run %s ended (%s) after %.1fs, %d tool calls",
445 self.run_id,
446 last.get("finish_reason") or last.get("message"),
447 time.monotonic() - started,
448 tools,
449 )
450
451 def add(self, item):
452 # Anything after an approval request means Hermes moved on (answered,
453 # or its own timeout).
454 settle_approvals(self)
455 if item["type"] == "approval":
456 item["code"] = open_approval(self, item)
457 with self.cond:
458 self.items.append(item)
459 self.cond.notify_all()
460 self.ask_unheard()
461
462 def attach(self):
463 with self.cond:
464 self.pages += 1
465
466 def detach(self):
467 with self.cond:
468 self.pages -= 1
469 self.ask_unheard()
470
471 def hang_up(self):
472 with self.cond:
473 self.hung_up = True
474 self.ask_unheard()
475
476 def ask_unheard(self):
477 """Send a pending approval to Matrix once no call can answer it."""
478 with self.cond:
479 if self.pages and not self.hung_up:
480 return
481 with APPROVALS_LOCK:
482 unasked = [
483 (code, item)
484 for code, (run, item) in APPROVALS.items()
485 if run is self and not item.get("asked")
486 ]
487 for _, item in unasked:
488 item["asked"] = True
489 for code, item in unasked:
490 ask_matrix(self.config, code, item)
491
492
493def open_approval(run, item):
494 with APPROVALS_LOCK:
495 code = secrets.token_hex(2)
496 while code in APPROVALS:
497 code = secrets.token_hex(2)
498 APPROVALS[code] = (run, item)
499 return code
500
501
502def settle_approvals(run):
503 with APPROVALS_LOCK:
504 for code in [code for code, (owner, _) in APPROVALS.items() if owner is run]:
505 del APPROVALS[code]
506
507
508def ask_matrix(config, code, item):
509 if not config.matrix_webhook:
510 log.info("approval %s has no listener and no Matrix webhook", code)
511 return
512 text = (
513 "Hermes needs your approval for work from a voice call: "
514 + (item.get("description") or "run a command")
515 )
516 if item.get("command"):
517 text += f"\n```\n{item['command']}\n```"
518 text += f"\nAnswer with `/assist approve {code}` or `/assist deny {code}`."
519 if item.get("timeout_s"):
520 text += f" Unanswered, it is denied after {round(item['timeout_s'] / 60)} minutes."
521 try:
522 notify_matrix(config, text)
523 log.info("approval %s asked in Matrix", code)
524 except (urllib.error.URLError, OSError) as error:
525 log.warning("asking approval %s in Matrix failed: %s", code, error)
526
527
528def stop_run(config, run_id):
529 """POST /v1/runs/{run_id}/stop; failures are only logged."""
530 try:
531 hermes_call(config, "POST", f"/v1/runs/{run_id}/stop")
532 log.info("stopped run %s", run_id)
533 except (urllib.error.URLError, OSError, ValueError) as error:
534 log.warning("stopping run %s failed: %s", run_id, error)
535
536
537class Handler(BaseHTTPRequestHandler):
538 server_version = "hermes-voice"
539 config: Config
540 static: Path
541 access_token: "str | None" = None
542
543 def authorized(self, also_hermes_key=False):
544 """Check the page's bearer token (or, where allowed, the API server key
545 the hermes-voice plugin in the gateway holds); answer 401 when it is
546 wrong."""
547 if self.access_token is None:
548 return True
549 header = self.headers.get("Authorization", "")
550 scheme, _, token = header.partition(" ")
551 accepted = [self.access_token]
552 if also_hermes_key:
553 accepted.append(self.config.hermes_key)
554 if scheme == "Bearer" and any(
555 hmac.compare_digest(token.encode(), key.encode()) for key in accepted
556 ):
557 return True
558 self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "access token required"})
559 return False
560
561 def log_message(self, format, *args):
562 log.info("%s %s", self.address_string(), format % args)
563
564 def send_json(self, status, payload):
565 body = json.dumps(payload).encode()
566 self.send_response(status)
567 self.send_header("Content-Type", "application/json")
568 self.send_header("Content-Length", str(len(body)))
569 self.send_header("Cache-Control", "no-store")
570 self.end_headers()
571 self.wfile.write(body)
572
573 def read_json(self):
574 length = int(self.headers.get("Content-Length") or 0)
575 if length <= 0 or length > MAX_BODY:
576 raise ValueError("bad body length")
577 return json.loads(self.rfile.read(length))
578
579 def do_GET(self):
580 if self.path == "/config":
581 if not self.authorized():
582 return
583 self.send_json(
584 HTTPStatus.OK,
585 {
586 "model": self.config.model,
587 "voice": self.config.voice,
588 "voices": self.config.voices(),
589 "idle_seconds": self.config.idle_seconds,
590 },
591 )
592 return
593 names = {
594 "/": ("index.html", "text/html; charset=utf-8"),
595 "/app.js": ("app.js", "text/javascript; charset=utf-8"),
596 "/style.css": ("style.css", "text/css; charset=utf-8"),
597 }
598 if self.path not in names:
599 self.send_error(HTTPStatus.NOT_FOUND)
600 return
601 name, content_type = names[self.path]
602 body = (self.static / name).read_bytes()
603 self.send_response(HTTPStatus.OK)
604 self.send_header("Content-Type", content_type)
605 self.send_header("Content-Length", str(len(body)))
606 self.send_header("Cache-Control", "no-store")
607 self.end_headers()
608 self.wfile.write(body)
609
610 def do_POST(self):
611 if not self.authorized(also_hermes_key=self.path == "/approve"):
612 return
613 try:
614 request = self.read_json()
615 except ValueError:
616 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON body"})
617 return
618 if self.path == "/session":
619 self.handle_session(request)
620 elif self.path == "/delegate":
621 self.handle_delegate(request)
622 elif self.path == "/hangup":
623 self.handle_hangup(request)
624 elif self.path == "/approve":
625 self.handle_approve(request)
626 else:
627 self.send_error(HTTPStatus.NOT_FOUND)
628
629 def handle_session(self, request):
630 # The vendor's SDP parser needs the offer byte-exact (trailing CRLF too).
631 sdp = request.get("sdp")
632 if not isinstance(sdp, str) or not sdp.strip():
633 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "sdp offer required"})
634 return
635 try:
636 voice = request.get("voice") or self.config.voice
637 if voice not in self.config.voices():
638 self.send_json(HTTPStatus.BAD_REQUEST, {"error": f"unknown voice {voice!r}"})
639 return
640 result = create_live_session(
641 self.config, sdp, live_history(request.get("history") or []), voice
642 )
643 except urllib.error.HTTPError as error:
644 detail = error.read().decode("utf-8", "replace")[:800]
645 log.warning("live session creation failed: %s %s", error.code, detail)
646 self.send_json(
647 HTTPStatus.BAD_GATEWAY,
648 {"error": f"OpenAI rejected the session ({error.code}): {detail}"},
649 )
650 return
651 except (urllib.error.URLError, OSError, KeyError, ValueError) as error:
652 log.warning("live session creation failed: %s", error)
653 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"OpenAI: {error}"})
654 return
655 self.send_json(HTTPStatus.OK, result)
656
657 def handle_delegate(self, request):
658 hermes_session = request.get("hermes_session")
659 lines = request.get("lines")
660 if not isinstance(hermes_session, str) or not SESSION_ID_RE.match(hermes_session):
661 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid hermes_session"})
662 return
663 if not isinstance(lines, list) or not lines:
664 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "lines required"})
665 return
666 # A correction within the same call (the page says so) supersedes and
667 # stops the run it replaces; runs from earlier calls keep going.
668 if request.get("replace") is True:
669 with RUNS_LOCK:
670 previous = LATEST.pop(hermes_session, None)
671 if previous:
672 stop_run(self.config, previous.run_id)
673 try:
674 run = Run(
675 self.config,
676 hermes_session,
677 hermes_start_run(self.config, hermes_session, lines),
678 )
679 except urllib.error.HTTPError as error:
680 detail = error.read().decode("utf-8", "replace")[:400]
681 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes {error.code}: {detail}"})
682 return
683 except (urllib.error.URLError, OSError, ValueError, KeyError) as error:
684 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes unreachable: {error}"})
685 return
686 self.send_response(HTTPStatus.OK)
687 self.send_header("Content-Type", "application/x-ndjson")
688 self.send_header("Cache-Control", "no-store")
689 self.send_header("X-Accel-Buffering", "no")
690 self.end_headers()
691 run.attach()
692 try:
693 sent = 0
694 while True:
695 with run.cond:
696 if sent == len(run.items) and not run.finished:
697 run.cond.wait(PAGE_KEEPALIVE_SECONDS)
698 new = run.items[sent:]
699 sent += len(new)
700 done = run.finished and sent == len(run.items)
701 for item in new:
702 self.write_line(item)
703 if done:
704 break
705 if not new:
706 self.wfile.write(b"\n")
707 self.wfile.flush()
708 except (BrokenPipeError, ConnectionResetError):
709 # Only the page is gone: this server keeps following the run.
710 log.info("page left run %s", run.run_id)
711 finally:
712 run.detach()
713
714 def handle_hangup(self, request):
715 """The call ended. No page hears this conversation's runs any more, and
716 what the user said since the last delegation becomes its own run."""
717 hermes_session = request.get("hermes_session")
718 lines = request.get("lines") or []
719 if (
720 not isinstance(hermes_session, str)
721 or not SESSION_ID_RE.match(hermes_session)
722 or not isinstance(lines, list)
723 ):
724 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid hang-up"})
725 return
726 with RUNS_LOCK:
727 LATEST.pop(hermes_session, None)
728 runs = [run for run in RUNS.values() if run.hermes_session == hermes_session]
729 for run in runs:
730 run.hang_up()
731 if not any(line.get("speaker") == "user" for line in lines):
732 self.send_json(HTTPStatus.OK, {"run_id": None})
733 return
734 try:
735 run_id = hermes_start_run(self.config, hermes_session, lines, hung_up=True)
736 except urllib.error.HTTPError as error:
737 detail = error.read().decode("utf-8", "replace")[:400]
738 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes {error.code}: {detail}"})
739 return
740 except (urllib.error.URLError, OSError, ValueError, KeyError) as error:
741 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes unreachable: {error}"})
742 return
743 Run(self.config, hermes_session, run_id, hung_up=True)
744 log.info("hang-up of %s left a request: run %s", hermes_session, run_id)
745 self.send_json(HTTPStatus.OK, {"run_id": run_id})
746
747 def handle_approve(self, request):
748 code = request.get("code")
749 choice = request.get("choice")
750 # No "always": a permanent allowlist entry is not a voice decision.
751 if choice not in ("once", "session", "deny") or not (
752 code is None or isinstance(code, str)
753 ):
754 self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid approval answer"})
755 return
756 with APPROVALS_LOCK:
757 if code is None and len(APPROVALS) == 1:
758 code = next(iter(APPROVALS))
759 pending = APPROVALS.get(code)
760 if pending is None:
761 self.send_json(
762 HTTPStatus.CONFLICT,
763 {"error": f"no pending approval {code}" if code else "no single pending approval"},
764 )
765 return
766 run, item = pending
767 try:
768 result = hermes_approve(self.config, run.run_id, item, choice)
769 except urllib.error.HTTPError as error:
770 detail = error.read().decode("utf-8", "replace")[:400]
771 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes {error.code}: {detail}"})
772 return
773 except (urllib.error.URLError, OSError, ValueError) as error:
774 self.send_json(HTTPStatus.BAD_GATEWAY, {"error": f"Hermes unreachable: {error}"})
775 return
776 with APPROVALS_LOCK:
777 APPROVALS.pop(code, None)
778 if result.get("status") == "already_resolved":
779 self.send_json(HTTPStatus.CONFLICT, {"error": f"approval {code} was already answered"})
780 return
781 log.info("approval %s (run %s): %s %s", code, run.run_id, choice, result.get("status"))
782 self.send_json(
783 HTTPStatus.OK,
784 {"code": code, "choice": choice, "description": item.get("description")},
785 )
786
787 def write_line(self, item):
788 self.wfile.write(json.dumps(item).encode() + b"\n")
789 self.wfile.flush()
790
791
792def main():
793 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
794 parser.add_argument("--listen", default="127.0.0.1:8790", help="host:port to serve on")
795 parser.add_argument("--hermes-url", required=True, help="Hermes API server base URL")
796 parser.add_argument("--hermes-key-file", required=True, help="file holding API_SERVER_KEY")
797 parser.add_argument("--openai-key-file", required=True, help="file holding the OpenAI API key")
798 parser.add_argument("--access-token-file", help="file holding the token the page must present")
799 parser.add_argument(
800 "--matrix-webhook",
801 help="Hermes webhook route (deliver_only to Matrix, signed with API_SERVER_KEY) that "
802 "receives approvals no call can answer",
803 )
804 parser.add_argument("--config", help="JSON file: voice, model, idle_seconds, glossary, instructions")
805 parser.add_argument(
806 "--static",
807 default=os.environ.get("HERMES_VOICE_STATIC"),
808 help="directory with index.html, app.js, style.css (default: $HERMES_VOICE_STATIC)",
809 )
810 args = parser.parse_args()
811 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
812
813 host, _, port = args.listen.rpartition(":")
814 host = host.strip("[]")
815 if args.access_token_file is None and host not in ("127.0.0.1", "::1", "localhost"):
816 parser.error("--access-token-file is required when not listening on loopback")
817
818 raw = json.loads(Path(args.config).read_text()) if args.config else {}
819 config = Config(
820 raw,
821 openai_key=Path(args.openai_key_file).read_text().strip(),
822 hermes_url=args.hermes_url,
823 hermes_key=Path(args.hermes_key_file).read_text().strip(),
824 matrix_webhook=args.matrix_webhook,
825 )
826 Handler.config = config
827 Handler.static = Path(args.static)
828 if args.access_token_file:
829 Handler.access_token = Path(args.access_token_file).read_text().strip()
830 server_class = ThreadingHTTPServer
831 if ":" in host:
832 # "::" is dual-stack on Linux (bindv6only=0).
833 server_class = type("Server6", (ThreadingHTTPServer,), {"address_family": socket.AF_INET6})
834 server = server_class((host, int(port)), Handler)
835 server.daemon_threads = True
836 log.info("serving on http://%s", args.listen)
837 server.serve_forever()
838
839
840if __name__ == "__main__":
841 main()