forked from nesquena/hermes-webui
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
749 lines (669 loc) · 29 KB
/
Copy pathserver.py
File metadata and controls
749 lines (669 loc) · 29 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
"""Hermes Web UI server entry point."""
import logging
import os
import re
import signal
import socket
import ssl
import sys
import threading
import time
import traceback
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
# Ignore SIGPIPE so a dropped client only aborts that write, not the whole WebUI process.
_SIGPIPE = getattr(signal, "SIGPIPE", None)
if _SIGPIPE is not None:
signal.signal(_SIGPIPE, signal.SIG_IGN)
# Test-mode network isolation keeps subprocess-backed tests hermetic.
if os.environ.get("HERMES_WEBUI_TEST_NETWORK_BLOCK", "").strip() in ("1", "true", "yes"):
_REAL_CREATE_CONN = socket.create_connection
_REAL_SOCK_CONNECT = socket.socket.connect
import re as _re
def _re_match_unique_local_ipv6(h):
"""Match IPv6 fc00::/7 without catching similar-looking hostnames."""
return bool(_re.match(r"^f[cd][0-9a-f]{0,2}:", h))
def _addr_is_local(host):
if not isinstance(host, str):
return False
h = host.strip().lower()
if not h:
return False
if h in ("::1", "0:0:0:0:0:0:0:1") or h.startswith("fe80:") or _re_match_unique_local_ipv6(h):
return True
if h == "localhost" or h.endswith(".localhost"):
return True
if h.endswith(".local") or h.endswith(".test") or h.endswith(".invalid"):
return True
if h == "example.com" or h.endswith(".example.com"):
return True
if h == "example.net" or h.endswith(".example.net"):
return True
if h == "example.org" or h.endswith(".example.org"):
return True
if h.endswith(".example"):
return True
if h and h[0].isdigit() and h.count(".") == 3:
try:
o1, o2, o3, o4 = [int(p) for p in h.split(".")]
except ValueError:
return False
if o1 == 127:
return True
if o1 == 10:
return True
if o1 == 192 and o2 == 168:
return True
if o1 == 172 and 16 <= o2 <= 31:
return True
if o1 == 169 and o2 == 254:
return True
if o1 == 203 and o2 == 0 and o3 == 113:
return True
return False
def _blocked_create_connection(address, *a, **kw):
try:
host = address[0]
except (TypeError, IndexError):
host = ""
if _addr_is_local(host):
return _REAL_CREATE_CONN(address, *a, **kw)
raise OSError(
f"hermes test network isolation (server.py): outbound to {address!r} blocked"
)
def _blocked_socket_connect(self, address):
try:
host = address[0]
except (TypeError, IndexError):
host = ""
if _addr_is_local(host):
return _REAL_SOCK_CONNECT(self, address)
raise OSError(
f"hermes test network isolation (server.py): socket.connect to {address!r} blocked"
)
socket.create_connection = _blocked_create_connection
socket.socket.connect = _blocked_socket_connect
try:
import resource
except ImportError: # pragma: no cover - resource is Unix-only
resource = None
from urllib.parse import urlparse
logger = logging.getLogger(__name__)
from api.auth import check_auth
from api.config import HOST, PORT, STATE_DIR, SESSION_DIR, DEFAULT_WORKSPACE
from api.helpers import (
j,
get_profile_cookie,
_build_csp_report_only_policy,
_CLIENT_DISCONNECT_ERRORS,
)
from api.profiles import set_request_profile, clear_request_profile
from api.routes import handle_delete, handle_get, handle_patch, handle_post, handle_put, apply_cors_preflight_headers
from api.startup import auto_install_agent_deps, fix_credential_permissions
from api.updates import WEBUI_VERSION
from api.crash_visibility import install_crash_visibility
class QuietHTTPServer(ThreadingHTTPServer):
"""Custom HTTP server that silently handles common network errors."""
daemon_threads = True
request_queue_size = 64
max_request_workers = 128
max_overflow_reject_workers = 16
_OVERFLOW_RESPONSE = (
b"HTTP/1.1 503 Service Unavailable\r\n"
b"Connection: close\r\n"
b"Content-Length: 0\r\n"
b"\r\n"
)
def __init__(self, *args, **kwargs):
server_address = args[0] if args else kwargs.get('server_address', None)
if server_address and ':' in server_address[0]:
self.address_family = socket.AF_INET6
self.ssl_context: object | None = None
super().__init__(*args, **kwargs)
self._request_worker_slots = threading.BoundedSemaphore(self.max_request_workers)
self._overflow_reject_slots = threading.BoundedSemaphore(self.max_overflow_reject_workers)
self.accept_loop_requests_total = 0
self.accept_loop_last_request_at = 0.0
def server_bind(self):
if sys.platform == 'win32':
self.allow_reuse_address = False
SO_EXCLUSIVEADDRUSE = getattr(socket, 'SO_EXCLUSIVEADDRUSE', -5)
self.socket.setsockopt(socket.SOL_SOCKET, SO_EXCLUSIVEADDRUSE, 1)
# Retry bind on Windows to handle the case where a previous
# process (e.g. during self-update) is still releasing the port.
# The old process calls os._exit(0) which starts tearing down
# its socket, but with SO_EXCLUSIVEADDRUSE the OS blocks new
# binds until the teardown completes. Retry for up to 10 s.
max_retries = 20
retry_delay = 0.5
for attempt in range(max_retries):
try:
super().server_bind()
return
except OSError as e:
if e.winerror == 10048 and attempt < max_retries - 1: # WSAEADDRINUSE
time.sleep(retry_delay)
else:
raise
else:
super().server_bind()
def get_request(self):
"""Accept raw sockets and defer TLS handshake work to request threads."""
request, client_address = self.socket.accept()
ssl_context = getattr(self, "ssl_context", None)
if ssl_context is None:
return request, client_address
try:
tls_request = ssl_context.wrap_socket(
request,
server_side=True,
do_handshake_on_connect=False,
)
except Exception:
request.close()
raise
return tls_request, client_address
def _handle_request_noblock(self):
"""Record accept-loop progress before dispatching a request handler."""
self.accept_loop_requests_total += 1
self.accept_loop_last_request_at = time.time()
return super()._handle_request_noblock()
def _close_request_quietly(self, request) -> None:
try:
request.close()
except Exception:
pass
def _drain_request_input_nonblocking(self, request) -> None:
# Read through the current header block before replying so Windows
# doesn't reset the socket when we close with unread input.
deadline = time.monotonic() + 0.05
buffered = bytearray()
header_terminator = b"\r\n\r\n"
max_bytes = 65536
try:
timeout = request.gettimeout()
except Exception:
timeout = None
try:
while len(buffered) < max_bytes and header_terminator not in buffered:
wait = deadline - time.monotonic()
if wait <= 0:
break
try:
request.settimeout(wait)
chunk = request.recv(min(4096, max_bytes - len(buffered)))
except (BlockingIOError, InterruptedError, TimeoutError, socket.timeout):
break
except OSError:
break
if not chunk:
break
buffered.extend(chunk)
try:
request.shutdown(socket.SHUT_RD)
except OSError:
pass
finally:
try:
request.settimeout(timeout)
except Exception:
pass
def _reject_overflow_request(self, request) -> None:
if getattr(self, "ssl_context", None) is not None:
self._close_request_quietly(request)
return
if not self._overflow_reject_slots.acquire(blocking=False):
self._close_request_quietly(request)
return
try:
threading.Thread(
target=self._reject_overflow_request_worker,
args=(request,),
daemon=True,
).start()
except Exception:
self._overflow_reject_slots.release()
self._close_request_quietly(request)
def _reject_overflow_request_worker(self, request) -> None:
try:
self._drain_request_input_nonblocking(request)
try:
request.sendall(self._OVERFLOW_RESPONSE)
try:
request.shutdown(socket.SHUT_WR)
except Exception:
pass
except Exception:
pass
finally:
self._close_request_quietly(request)
self._overflow_reject_slots.release()
def process_request(self, request, client_address):
if not self._request_worker_slots.acquire(blocking=False):
self._reject_overflow_request(request)
return
try:
return super().process_request(request, client_address)
except Exception:
self._request_worker_slots.release()
self._close_request_quietly(request)
raise
def process_request_thread(self, request, client_address):
try:
return super().process_request_thread(request, client_address)
finally:
self._request_worker_slots.release()
def handle_error(self, request, client_address):
"""Suppress logging for common client disconnect errors."""
exc_type, exc_value, _ = sys.exc_info()
if exc_type in (
ConnectionResetError, BrokenPipeError, ConnectionAbortedError,
TimeoutError, ssl.SSLError, ssl.SSLEOFError,
):
return
if issubclass(exc_type, OSError):
if getattr(exc_value, 'errno', None) in (32, 54, 104, 110): # EPIPE, ECONNRESET, ETIMEDOUT
return
super().handle_error(request, client_address)
class Handler(BaseHTTPRequestHandler):
# HTTP/1.1 keep-alive stays on, so every response must declare framing.
protocol_version = "HTTP/1.1"
timeout = 30 # seconds — kills idle/incomplete connections to prevent thread exhaustion
def setup(self):
"""Set socket options for each accepted connection."""
super().setup()
try:
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
except OSError:
pass
try:
self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
except OSError:
pass
if hasattr(socket, 'TCP_KEEPIDLE'): # Linux
try:
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 10)
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 5)
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)
except OSError:
pass
elif hasattr(socket, 'TCP_KEEPALIVE'): # macOS
try:
self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPALIVE, 10)
except OSError:
pass
_ver_suffix = WEBUI_VERSION.removeprefix('v')
server_version = ('HermesWebUI/' + _ver_suffix) if _ver_suffix != 'unknown' else 'HermesWebUI'
_CSP_REPORT_TO = '{"group":"csp-endpoint","max_age":10886400,"endpoints":[{"url":"/api/csp-report"}]}'
@classmethod
def csp_report_only_policy(cls, extra_connect_src=None, extra_frame_src=None) -> str:
return _build_csp_report_only_policy(extra_connect_src, extra_frame_src)
def end_headers(self) -> None:
extra_connect_src = getattr(self, "_csp_extra_connect_src", None)
extra_frame_src = getattr(self, "_csp_extra_frame_src", None)
self.send_header("Content-Security-Policy-Report-Only", self.csp_report_only_policy(extra_connect_src, extra_frame_src))
self.send_header("Report-To", self._CSP_REPORT_TO)
super().end_headers()
def log_message(self, fmt, *args): pass # suppress default Apache-style log
@staticmethod
def _safe_webui_print(message: str) -> None:
"""Emit a request log line without letting logging break responses."""
try:
print(message, flush=True)
except Exception:
pass
def log_request(self, code: str='-', size: str='-') -> None:
"""Structured JSON logs for each request."""
import json as _json
duration_ms = round((time.time() - getattr(self, '_req_t0', time.time())) * 1000, 1)
remote = '-'
try:
if getattr(self, 'client_address', None):
remote = str(self.client_address[0])
except Exception:
remote = '-'
forwarded_for = None
try:
forwarded_for = (self.headers.get('X-Forwarded-For') or '').split(',')[0].strip() or None
except Exception:
forwarded_for = None
record_data = {
'ts': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
'remote': remote,
'method': getattr(self, 'command', None) or '-',
'path': getattr(self, 'path', None) or '-',
'status': int(code) if str(code).isdigit() else code,
'ms': duration_ms,
}
if forwarded_for:
record_data['forwarded_for'] = forwarded_for
record = _json.dumps(record_data)
self._safe_webui_print(f'[webui] {record}')
def do_GET(self) -> None:
self._req_t0 = time.time()
cookie_profile = get_profile_cookie(self)
if cookie_profile:
set_request_profile(cookie_profile)
try:
parsed = urlparse(self.path)
if not check_auth(self, parsed): return
result = handle_get(self, parsed)
if result is False:
return j(self, {'error': 'not found'}, status=404)
except _CLIENT_DISCONNECT_ERRORS:
# Expected disconnect path; do not convert it into a misleading server 500.
return
except Exception:
self._safe_webui_print(f'[webui] ERROR {self.command} {self.path}\n' + traceback.format_exc())
try:
j(self, {'error': 'Internal server error'}, status=500)
except _CLIENT_DISCONNECT_ERRORS:
pass
except Exception:
self._safe_webui_print(traceback.format_exc())
finally:
clear_request_profile()
def _handle_write(self, route_func) -> None:
self._req_t0 = time.time()
cookie_profile = get_profile_cookie(self)
if cookie_profile:
set_request_profile(cookie_profile)
try:
parsed = urlparse(self.path)
_is_csp_report_post = (
parsed.path == "/api/csp-report" and self.command == "POST"
)
if not _is_csp_report_post and not check_auth(self, parsed): return
result = route_func(self, parsed)
if result is False:
return j(self, {'error': 'not found'}, status=404)
except _CLIENT_DISCONNECT_ERRORS:
# Expected disconnect path; do not convert it into a misleading server 500.
return
except Exception:
self._safe_webui_print(f'[webui] ERROR {self.command} {self.path}\n' + traceback.format_exc())
try:
j(self, {'error': 'Internal server error'}, status=500)
except _CLIENT_DISCONNECT_ERRORS:
pass
except Exception:
self._safe_webui_print(traceback.format_exc())
finally:
clear_request_profile()
def do_POST(self) -> None:
self._handle_write(handle_post)
def do_PUT(self) -> None:
self._handle_write(handle_put)
def do_PATCH(self) -> None:
self._handle_write(handle_patch)
def do_OPTIONS(self) -> None:
"""Handle CORS preflight requests (headers emitted by api.routes)."""
self._req_t0 = time.time()
self.send_response(200)
apply_cors_preflight_headers(self)
# Frame the empty preflight: without Content-Length an HTTP/1.1 keep-alive
# 200 is read-until-close, hanging the client until the 30s timeout.
self.send_header("Content-Length", "0")
self.end_headers()
def do_DELETE(self) -> None:
self._handle_write(handle_delete)
def _raise_fd_soft_limit(target: int = 4096) -> dict:
"""Best-effort raise of RLIMIT_NOFILE for persistent WebUI hosts."""
if resource is None:
return {"status": "unsupported"}
try:
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
except Exception as exc:
return {"status": "error", "error": str(exc)}
desired = int(target)
if hard not in (-1, getattr(resource, "RLIM_INFINITY", object())):
desired = min(desired, int(hard))
if soft >= desired:
return {"status": "unchanged", "soft": soft, "hard": hard}
try:
resource.setrlimit(resource.RLIMIT_NOFILE, (desired, hard))
except Exception as exc:
return {"status": "error", "soft": soft, "hard": hard, "error": str(exc)}
return {"status": "raised", "soft": desired, "hard": hard, "previous_soft": soft}
_SHUTDOWN_AUDIT_LOGGED = False
_SHUTDOWN_LOG_VALUE_RE = re.compile(r"[\x00-\x1f\x7f]+")
def _shutdown_log_value(value, *, default: str = "unknown", max_len: int = 160) -> str:
"""Return a bounded single-line value safe for shutdown diagnostics."""
if value is None:
return default
try:
text = str(value)
except Exception:
return default
text = _SHUTDOWN_LOG_VALUE_RE.sub("?", text).strip()
if not text:
return default
if len(text) > max_len:
text = f"{text[:max_len]}…"
return text
def _log_shutdown_audit(reason: str = "serve_forever_exit") -> None:
"""Log runtime context when the WebUI server is exiting."""
global _SHUTDOWN_AUDIT_LOGGED
if _SHUTDOWN_AUDIT_LOGGED:
return
active_sessions = []
try:
from api.models import LOCK, SESSIONS
with LOCK:
session_items = list(SESSIONS.items())
for sid, session in session_items:
stream_id = getattr(session, "active_stream_id", None)
if stream_id:
pending = bool(getattr(session, "pending_user_message", None))
active_sessions.append(
"sid=%s stream=%s pending=%s"
% (
_shutdown_log_value(sid),
_shutdown_log_value(stream_id),
pending,
)
)
except Exception:
logger.debug("Failed to collect active-session shutdown audit state", exc_info=True)
_SHUTDOWN_AUDIT_LOGGED = True
logger.info(
"[shutdown-audit] reason=%s pid=%s thread=%s(%s) active_sessions=[%s]",
_shutdown_log_value(reason),
os.getpid(),
_shutdown_log_value(threading.current_thread().name),
threading.current_thread().ident,
"; ".join(active_sessions) if active_sessions else "none",
)
def _abort_if_already_serving(host: str, port: int) -> None:
"""Refuse to start if a live HTTP server is already responding on this port."""
probe_host = '127.0.0.1' if host in ('0.0.0.0', '', '::') else host
try:
with socket.create_connection((probe_host, port), timeout=2) as s:
s.sendall(b'GET /health HTTP/1.0\r\nHost: localhost\r\n\r\n')
s.settimeout(2)
data = s.recv(512)
if data:
print(
f'[!!] FATAL: Another server is already responding on'
f' {probe_host}:{port}. Stop the existing instance first.',
flush=True,
)
sys.exit(1)
except (ConnectionRefusedError, ConnectionResetError, OSError, socket.timeout):
pass
def main() -> None:
from api.config import print_startup_config, verify_hermes_imports, _HERMES_FOUND
# Crash visibility FIRST (issue #4633): enable faulthandler + excepthooks +
# exit audit before any heavy startup work so a native crash or a daemon /
# handler-thread exception during startup or serving produces a diagnostic
# instead of a silent death. The paired memory root-cause is #4765.
install_crash_visibility()
print_startup_config()
fd_limit = _raise_fd_soft_limit()
if fd_limit.get("status") == "raised":
print(
f"[ok] Raised file descriptor soft limit "
f"{fd_limit.get('previous_soft')} -> {fd_limit.get('soft')}",
flush=True,
)
elif fd_limit.get("status") == "error":
print(f"[!!] WARNING: Could not raise file descriptor limit: {fd_limit.get('error')}", flush=True)
fix_credential_permissions()
try:
from api.models import _active_state_db_path
from api.session_recovery import recover_all_sessions_on_startup
result = recover_all_sessions_on_startup(
SESSION_DIR,
rebuild_index=True,
state_db_path=_active_state_db_path(),
)
if result.get("restored"):
print(f"[recovery] Restored {result['restored']}/{result['scanned']} sessions from .bak (see #1558).", flush=True)
except Exception as exc:
# Recovery is best-effort; never block server startup.
print(f"[recovery] startup recovery failed: {exc}", flush=True)
within_container = False
try:
with open('/.within_container', 'r') as f:
within_container = True
except FileNotFoundError:
pass
if within_container:
print('[ok] Running within container.', flush=True)
# Security: warn if binding non-loopback without authentication
from api.auth import get_oidc_startup_warning, is_auth_enabled
if HOST not in ('127.0.0.1', '::1', 'localhost') and not is_auth_enabled():
print(f'[!!] WARNING: Binding to {HOST} with NO PASSWORD SET.', flush=True)
print(f' Anyone on the network can access your filesystem and agent.', flush=True)
print(f' Set a password via Settings or HERMES_WEBUI_PASSWORD env var.', flush=True)
print(f' To suppress: bind to 127.0.0.1 or set a password.', flush=True)
if within_container:
print(f' Note: You are running within a container, must bind to 0.0.0.0 (IPv4) or :: (IPv6) to publish the port.', flush=True)
elif not is_auth_enabled():
print(f' [tip] No password set. Any process on this machine can read sessions', flush=True)
print(f' and memory via the local API. Set HERMES_WEBUI_PASSWORD to', flush=True)
print(f' enable authentication.', flush=True)
oidc_startup_warning = get_oidc_startup_warning()
if oidc_startup_warning:
print(f'[!!] WARNING: {oidc_startup_warning}', flush=True)
ok, missing, errors = verify_hermes_imports()
if not ok and _HERMES_FOUND:
print(f'[!!] Warning: Hermes agent found but missing modules: {missing}', flush=True)
for mod, err in errors.items():
print(f' {mod}: {err}', flush=True)
print(' Attempting to install missing dependencies from agent requirements.txt...', flush=True)
auto_install_agent_deps()
ok, missing, errors = verify_hermes_imports()
if not ok:
print(f'[!!] Still missing after install attempt: {missing}', flush=True)
for mod, err in errors.items():
print(f' {mod}: {err}', flush=True)
print(' Agent features may not work correctly.', flush=True)
else:
print('[ok] Agent dependencies installed successfully.', flush=True)
STATE_DIR.mkdir(parents=True, exist_ok=True)
SESSION_DIR.mkdir(parents=True, exist_ok=True)
DEFAULT_WORKSPACE.mkdir(parents=True, exist_ok=True)
try:
from api.gateway_watcher import start_watcher
def _start_watcher_safe():
try:
start_watcher()
except Exception as e:
print(f'[!!] WARNING: Gateway watcher failed to start: {e}', flush=True)
t = threading.Thread(target=_start_watcher_safe, daemon=True)
t.start()
t.join(timeout=5)
if t.is_alive():
print('[tip] Gateway watcher still initializing (non-blocking)', flush=True)
except Exception as e:
print(f'[!!] WARNING: Gateway watcher failed to start: {e}', flush=True)
try:
from api.background_process import start_drain_thread
if start_drain_thread():
print('[ok] bg_task_complete drain thread started', flush=True)
except Exception as e:
print(f'[!!] WARNING: bg_task_complete drain failed to start: {e}', flush=True)
try:
from api.background_process import start_session_channel_reaper
if start_session_channel_reaper():
print('[ok] SessionChannel reaper thread started', flush=True)
except Exception as e:
print(f'[!!] WARNING: SessionChannel reaper failed to start: {e}', flush=True)
try:
from api.plugins import load_plugins
load_plugins()
except Exception as e:
print(f'[!!] WARNING: Plugin loading failed: {e}', flush=True)
_abort_if_already_serving(HOST, PORT)
httpd = QuietHTTPServer((HOST, PORT), Handler)
from api.config import TLS_ENABLED, TLS_CERT, TLS_KEY
scheme = 'https' if TLS_ENABLED else 'http'
if TLS_ENABLED:
try:
import ssl
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
ctx.load_cert_chain(TLS_CERT, TLS_KEY)
httpd.ssl_context = ctx
print(f' TLS enabled: cert={TLS_CERT}, key={TLS_KEY}', flush=True)
except Exception as e:
print(f'[!!] WARNING: TLS setup failed ({e}), falling back to HTTP', flush=True)
scheme = 'http'
print(f' Hermes Web UI listening on {scheme}://{HOST}:{PORT}', flush=True)
if HOST in ('127.0.0.1', '::1') or within_container:
print(f' Remote access: ssh -N -L {PORT}:127.0.0.1:{PORT} <user>@<your-server>', flush=True)
print(f' Then open: {scheme}://localhost:{PORT}', flush=True)
print('', flush=True)
# ctl.sh stops the WebUI with SIGTERM. Python's default SIGTERM handler
# terminates the process WITHOUT unwinding the try/finally around
# serve_forever(), so drain_all_on_shutdown() (which flushes in-flight
# fire-and-forget memory commits) would never run on the normal managed
# stop. Install a handler that requests an orderly shutdown so
# serve_forever() returns and the existing `finally` block drains cleanly.
#
# httpd.shutdown() blocks until serve_forever() has exited and MUST NOT be
# called from the thread running serve_forever() (it would deadlock), so we
# dispatch it from a short-lived helper thread. The handler is idempotent
# and guards against double-shutdown (e.g. repeated SIGTERM/SIGINT).
_shutdown_requested = threading.Event()
def _request_shutdown(signum, _frame):
if _shutdown_requested.is_set():
return
_shutdown_requested.set()
threading.Thread(
target=httpd.shutdown,
name="webui-sigterm-shutdown",
daemon=True,
).start()
try:
signal.signal(signal.SIGTERM, _request_shutdown)
except (ValueError, OSError):
# Not on the main thread (e.g. embedded/test harness); skip handler.
logger.debug("Could not install SIGTERM handler", exc_info=True)
try:
httpd.serve_forever()
finally:
httpd.server_close()
_log_shutdown_audit()
try:
from api.gateway_watcher import stop_watcher
stop_watcher()
except Exception:
logger.debug("Failed to stop gateway watcher during shutdown")
try:
from api.session_lifecycle import drain_all_on_shutdown
drain_all_on_shutdown()
except Exception:
logger.debug("Failed to drain lifecycle on shutdown", exc_info=True)
try:
from api.background_process import stop_drain_thread
stop_drain_thread()
except Exception:
logger.debug("Failed to stop bg_task_complete drain thread during shutdown", exc_info=True)
try:
from api.background_process import stop_session_channel_reaper
stop_session_channel_reaper()
except Exception:
logger.debug("Failed to stop SessionChannel reaper during shutdown", exc_info=True)
if __name__ == '__main__':
main()