{{ message }}
Repository navigation
Expand file tree
/
Copy pathhermes_logging.py
More file actions
1043 lines (897 loc) · 45.8 KB
/
Copy pathhermes_logging.py
File metadata and controls
1043 lines (897 loc) · 45.8 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
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Centralized logging setup for Hermes Agent.
Log files: agent.log (INFO+, everything), errors.log (WARNING+), gateway.log (INFO+,
gateway components; ``mode="gateway"``), gui.log (INFO+, dashboard/TUI-gateway;
``mode="gui"``). All are rotating files driven through one async queue and formatted
with ``RedactingFormatter`` so secrets never reach disk.
"""
import atexit
import contextlib
import copy
import io
import logging
import os
import queue
import sys
import threading
from logging.handlers import QueueHandler, QueueListener
from pathlib import Path
from time import monotonic as _monotonic
from typing import Optional, Sequence
from hermes_constants import (
get_config_path, get_hermes_home, mkdir_under_hermes_home, named_profile_is_deleted,
)
# setup_logging() is idempotent: a second call is a no-op unless ``force=True``.
_logging_initialized = False
# True only when CLH was rejected at import because portalocker cannot take a
# lock on this Windows box; file handlers then use stdlib rotation (rollover
# disabled — see the module-header comment) and setup_logging() warns once.
_WINDOWS_CLH_FALLBACK = False
_WINDOWS_CLH_FALLBACK_REASON = ""
_fallback_warned = False
def _portalocker_probe() -> bool:
"""Return True when portalocker can actually take a lock on this box.
concurrent-log-handler locks every write through portalocker, which on
Windows instantiates Win32Locker and imports pywintypes. Bundled payloads
have shipped with that import broken (the venv's .pth files were never
processed), and portalocker 3.x gives no msvcrt fallback — every emit then
dies with the ImportError, CLH retries 20x, and the suppressed "Cannot
acquire lock" RuntimeError below hides it completely. Probe a scratch file
once at import so we can fall back to stdlib rotation instead of silently
dropping every record. No-op (True) off Windows, where stdlib is in use.
"""
global _WINDOWS_CLH_FALLBACK_REASON
if sys.platform != "win32":
return True
try:
import portalocker
import tempfile
except Exception as exc:
_WINDOWS_CLH_FALLBACK_REASON = repr(exc)
return False
fd, path = tempfile.mkstemp(prefix="hermes-portalocker-")
try:
with os.fdopen(fd, "r+b") as stream:
portalocker.lock(stream, portalocker.LOCK_EX)
portalocker.unlock(stream)
except Exception as exc:
_WINDOWS_CLH_FALLBACK_REASON = repr(exc)
return False
finally:
try:
os.unlink(path)
except OSError:
pass
return True
# Windows-ONLY swap (#44873): stdlib ``RotatingFileHandler.doRollover()`` calls
# ``os.rename()``, which fails with ``PermissionError [WinError 32]`` whenever
# another process holds an append handle on ``agent.log`` — essentially always
# in Hermes (TUI, gateway, hy_memory, MCP servers, CLI commands all log) —
# pinning the file at the size threshold and spamming stderr on every emit.
# ``concurrent-log-handler`` serializes rollover with a cross-process lock.
# POSIX keeps stdlib: renames of open files work, and managed mode (NixOS)
# relies on stdlib's exact ``_open()``/``doRollover()`` lifecycle for the
# 0660 chmod and eager file creation; CLH opens lazily and rotates differently.
if sys.platform == "win32":
if _portalocker_probe():
from concurrent_log_handler import (
ConcurrentRotatingFileHandler as RotatingFileHandler,
)
else:
# portalocker cannot take a lock on this box (typical cause: a sealed
# bundle whose venv never processed pywin32.pth, so `import pywintypes`
# fails and portalocker's Win32Locker has no msvcrt fallback). CLH
# would silently drop every record through the suppressed lock-timeout
# below; fall back to stdlib rotation instead. Rollover is disabled in
# the fallback: multi-process appends make Windows renames fail with
# WinError 32, the exact #44873 trap CLH exists to avoid.
from logging.handlers import RotatingFileHandler
_WINDOWS_CLH_FALLBACK = True
else:
from logging.handlers import RotatingFileHandler
# Thread-local per-conversation session context.
_session_context = threading.local()
# ``%(session_tag)s`` exists on every LogRecord via _install_session_record_factory().
_LOG_FORMAT = "%(asctime)s %(levelname)s%(session_tag)s %(name)s: %(message)s"
_LOG_FORMAT_VERBOSE = "%(asctime)s - %(name)s - %(levelname)s%(session_tag)s - %(message)s"
# The stdout stream _line_buffer_piped_stdout() already reconfigured: setup_logging runs on every
# AIAgent build (per message in the gateway), so later calls skip the flush + reconfigure.
_line_buffered_stdout = None
def _line_buffer_piped_stdout() -> None:
"""Best-effort line buffering for a piped stdout (#92281).
Python block-buffers stdout when it isn't a TTY, so an agent loop
driving ``print()`` into a supervisor's pipe delivers its output in
large delayed bursts — a headless run looks stalled for minutes while
the work is actually progressing (only stderr, which we already give
``line_buffering=True`` in ``_safe_stderr()``, shows up on time).
Reconfigure the interpreter's own stdout for line buffering when it is
piped; interactive TTY stdout is already line-buffered. ACP reaches this
too (its AIAgent calls ``setup_logging``) with stdout as its protocol
channel — harmless, since line buffering only adds flushes at newlines.
"""
global _line_buffered_stdout
stream = sys.stdout
if stream is None or stream is _line_buffered_stdout:
return
if getattr(stream, "line_buffering", False) is True:
_line_buffered_stdout = stream
return
reconfigure = getattr(stream, "reconfigure", None)
if reconfigure is None:
return
# A closed/detached stream raises ValueError / io.UnsupportedOperation (an OSError);
# buffering is observability, not correctness — never crash setup_logging over it.
with contextlib.suppress(OSError, ValueError):
if not stream.isatty():
reconfigure(line_buffering=True)
_line_buffered_stdout = stream
def _safe_stderr(): # type: ignore[return]
"""Return a stderr stream that tolerates Unicode on all platforms.
Wraps ``sys.stderr`` with ``errors='replace'`` so un-encodable characters become
``?`` instead of crashing the process.
"""
stream = sys.stderr
encoding = getattr(stream, "encoding", None) or "utf-8"
if encoding.lower().replace("-", "") in ("utf8", "utf8surrogateescape"):
return stream
try:
wrapped = io.TextIOWrapper(stream.buffer, encoding="utf-8", errors="replace", line_buffering=True)
# Prevent the wrapper from closing the underlying buffer when garbage-collected.
wrapped.close = lambda: None # type: ignore[assignment]
return wrapped
except Exception:
return stream # best-effort: no buffer / wrapping failed -> original stream
class _RedactingStdio:
"""Text stream proxy that redacts each write: a detached gateway's stdout/stderr IS a
persisted log (``logs/gateway-stdio.log``), and agent status lines (provider error echoes)
reach it through ``print``, never through ``RedactingFormatter``."""
def __init__(self, stream):
self._stream = stream
def write(self, text):
from agent.redact import redact_sensitive_text
return self._stream.write(redact_sensitive_text(text) if isinstance(text, str) else text)
def __getattr__(self, name):
return getattr(self._stream, name)
def redact_detached_stdio() -> None:
"""Route a detached process's stdout/stderr through secret redaction (idempotent)."""
for name in ("stdout", "stderr"):
stream = getattr(sys, name)
if stream is not None and not isinstance(stream, _RedactingStdio):
setattr(sys, name, _RedactingStdio(stream))
def _is_windows_concurrent_log_lock_timeout(exc: BaseException | None) -> bool:
"""True for concurrent-log-handler's Windows lock timeout.
Slash-command workers and the gateway share rotating files on Windows Desktop;
when another process holds the rollover lock too long CLH raises this
RuntimeError, which must not escape into Desktop chat output.
"""
return (
sys.platform == "win32"
and isinstance(exc, RuntimeError)
and "Cannot acquire lock after 20 attempts" in str(exc)
)
_windows_lock_timeout_warned = False
_windows_lock_timeout_warn_lock = threading.Lock()
def _warn_windows_lock_timeout_once() -> None:
"""Report a suppressed CLH lock timeout exactly once per process.
Every emit after the first failure raises the same RuntimeError, so the
warning must be one-shot or it would spam errors.log as badly as the
stderr noise it replaces. CLH does not chain the underlying cause (the
RuntimeError is raised outside the except, from the retry loop's else
clause), so the message cannot include it; a constantly repeating timeout
means file logging is degraded — the startup portalocker probe in this
module should have caught a dead portalocker and fallen back already.
"""
global _windows_lock_timeout_warned
with _windows_lock_timeout_warn_lock:
if _windows_lock_timeout_warned:
return
_windows_lock_timeout_warned = True
logging.getLogger("hermes_logging").warning(
"concurrent-log-handler timed out acquiring the cross-process log "
"lock; this and later records were dropped (the Desktop slash-worker "
"surface stays clean, but file logging is degraded)."
)
def _is_unavailable_log_stream(exc: BaseException | None) -> bool:
"""True when a file handler lost its backing stream during teardown or I/O."""
return (
(isinstance(exc, OSError) and exc.errno == 5)
or (isinstance(exc, ValueError) and "closed file" in str(exc).lower())
)
# Third-party loggers that are noisy at DEBUG/INFO level.
_NOISY_LOGGERS = (
"openai", "openai._base_client", "httpx", "httpcore", "asyncio", "hpack", "hpack.hpack",
"grpc", "modal", "urllib3", "urllib3.connectionpool", "websockets", "charset_normalizer",
"markdown_it",
)
def _quiet_noisy_loggers() -> None:
"""Pin noisy third-party loggers at WARNING."""
for name in _NOISY_LOGGERS:
logging.getLogger(name).setLevel(logging.WARNING)
def set_session_context(session_id: str) -> None:
"""Set the session ID for the current thread."""
_session_context.session_id = session_id
def clear_session_context() -> None:
"""Clear the session ID for the current thread."""
_session_context.session_id = None
def _install_session_record_factory() -> None:
"""Replace the global LogRecord factory with one that adds ``session_tag``.
Unlike a Filter, the record factory runs for EVERY record in the process (propagated
and third-party-handled ones included), so ``%(session_tag)s`` never KeyErrors.
Idempotent via a marker attribute.
"""
current_factory = logging.getLogRecordFactory()
if getattr(current_factory, "_hermes_session_injector", False):
return
def _session_record_factory(*args, **kwargs):
record = current_factory(*args, **kwargs)
sid = getattr(_session_context, "session_id", None)
record.session_tag = f" [{sid}]" if sid else "" # type: ignore[attr-defined]
# QueueListener formats on its own thread, after the profile-scoped
# ContextVar is gone; keep the resolved home on the record so a
# multiplex desktop ticker can route to the job owner's files (#97489).
try:
record.hermes_home = str(get_hermes_home().resolve()) # type: ignore[attr-defined]
except Exception:
record.hermes_home = "" # type: ignore[attr-defined]
return record
_session_record_factory._hermes_session_injector = True # type: ignore[attr-defined]
logging.setLogRecordFactory(_session_record_factory)
# Install on import so session_tag exists on all records even before setup_logging().
_install_session_record_factory()
class _ComponentFilter(logging.Filter):
"""Only pass records whose logger name starts with one of *prefixes*."""
def __init__(self, prefixes: Sequence[str]) -> None:
super().__init__()
self._prefixes = tuple(prefixes)
def filter(self, record: logging.LogRecord) -> bool:
return record.name.startswith(self._prefixes)
# Logger name prefixes per component; used by _ComponentFilter and ``hermes logs --component``.
COMPONENT_PREFIXES = {
# ``plugins.platforms``: messaging adapters that migrated out of
# ``gateway/platforms/`` into bundled plugins (#41112) are still gateway
# components and belong in gateway.log. ``agent.memory_caller_scope``: a multiplexed gateway's
# memory-provider profile refusals (#140058), which the operator reads in gateway.log.
"gateway": ("gateway", "hermes_plugins", "plugins.platforms", "agent.memory_caller_scope"),
"agent": ("agent", "run_agent", "model_tools", "batch_runner"),
"tools": ("tools",),
"cli": ("hermes_cli", "cli"),
"cron": ("cron",),
"gui": ("hermes_cli.web_server", "hermes_cli.pty_bridge", "hermes_cli.desktop", "tui_gateway", "uvicorn"),
}
def _known_log_homes() -> set[Path]:
"""Homes the queued file handlers already serve: static handlers by their file, routers by
their default home plus every profile home they route. Caller holds ``_queue_state_lock``."""
homes: set[Path] = set()
for handler in _queued_file_handlers:
if isinstance(handler, _ProfileRoutingFileHandler):
homes.add(handler._default_home)
homes.update(handler._profile_homes)
elif isinstance(handler, RotatingFileHandler):
try:
homes.add(Path(handler.baseFilename).resolve().parent.parent)
except (TypeError, ValueError, OSError):
continue
return homes
def _adopt_secondary_home(home: Path) -> bool:
"""Route *home*'s records to its own files when this process already logs for another home.
Enables profile routing for the union of homes (or widens the live routers); False when
*home* is the first home seen or is already served."""
try:
resolved = Path(home).expanduser().resolve()
except (TypeError, ValueError, OSError):
return False
with _queue_state_lock:
known = _known_log_homes()
if not known or resolved in known:
return False
return enable_profile_log_routing([*sorted(known), resolved])
def setup_logging(
*,
hermes_home: Optional[Path] = None,
log_level: Optional[str] = None,
max_size_mb: Optional[int] = None,
backup_count: Optional[int] = None,
mode: Optional[str] = None,
force: bool = False,
) -> Path:
"""Configure the Hermes logging subsystem; returns the ``logs/`` directory.
Safe to call multiple times; the second call is a no-op unless *force*. Level and
rotation defaults come from config.yaml ``logging.*``. ``mode="gateway"`` adds
``gateway.log`` and ``mode="gui"`` adds ``gui.log``.
"""
global _logging_initialized
global _fallback_warned
home = hermes_home or get_hermes_home()
log_dir = mkdir_under_hermes_home(home / "logs")
# Stdout is block-buffered when piped (no TTY); line-buffer it so a
# headless supervisor's log stream tracks the agent loop incrementally
# (#92281). Runs before the initialized check so every entry mode that
# reaches this function gets it; a no-op once this stdout is line-buffered.
_line_buffer_piped_stdout()
# A second Hermes home in a process that already logs for another one — a dashboard or
# ``hermes serve`` backend building agents for several profiles, a multiplexed gateway —
# gets routed by record home. Stacking another file handler here would hand it EVERY
# profile's records (the handlers carry no home filter), and a duplicate writer on top of
# an existing router.
if _adopt_secondary_home(home):
return log_dir
cfg_level, cfg_max_size, cfg_backup = _read_logging_config()
level_name = (log_level or cfg_level or "INFO").upper()
level = getattr(logging, level_name, logging.INFO)
max_bytes = (max_size_mb or cfg_max_size or 5) * 1024 * 1024
backups = backup_count or cfg_backup or 3
from agent.redact import RedactingFormatter # lazy: circular at module load
root = logging.getLogger()
# (filename, level, max_bytes, backup_count, component) — a component gates
# the file on ``mode`` and restricts it to that component's logger prefixes.
handler_specs = (
("agent.log", level, max_bytes, backups, None),
("errors.log", logging.WARNING, 2 * 1024 * 1024, 2, None),
("gateway.log", logging.INFO, 5 * 1024 * 1024, 3, "gateway"),
("gui.log", logging.INFO, 10 * 1024 * 1024, 5, "gui"),
)
for filename, lvl, size, count, component in handler_specs:
if component is not None and mode != component:
continue
_add_rotating_handler(
log_dir / filename, level=lvl, max_bytes=size, backup_count=count,
formatter=RedactingFormatter(_LOG_FORMAT),
log_filter=_ComponentFilter(COMPONENT_PREFIXES[component]) if component else None,
)
if _WINDOWS_CLH_FALLBACK and not _fallback_warned:
# One-shot, and the file handlers above are already live, so this lands
# in errors.log/agent.log — the fallback must never be invisible again.
_fallback_warned = True
logging.getLogger("hermes_logging").warning(
"concurrent-log-handler unavailable on this Windows install (%s); "
"file logging fell back to stdlib rotation without rollover.",
_WINDOWS_CLH_FALLBACK_REASON or "portalocker probe failed",
)
if _logging_initialized and not force:
return log_dir
# Root level must be low enough for the handlers to fire.
if root.level == logging.NOTSET or root.level > level:
root.setLevel(level)
_quiet_noisy_loggers()
_logging_initialized = True
return log_dir
def setup_verbose_logging() -> None:
"""Enable DEBUG-level console logging for ``--verbose`` / ``-v`` mode."""
from agent.redact import RedactingFormatter
root = logging.getLogger()
if any(getattr(h, "_hermes_verbose", False) for h in root.handlers):
return
handler = logging.StreamHandler(_safe_stderr())
handler.setLevel(logging.DEBUG)
handler.setFormatter(RedactingFormatter(_LOG_FORMAT_VERBOSE, datefmt="%H:%M:%S"))
handler._hermes_verbose = True # type: ignore[attr-defined]
root.addHandler(handler)
if root.level > logging.DEBUG:
root.setLevel(logging.DEBUG)
_quiet_noisy_loggers()
# rex-deploy at INFO for sandbox status.
logging.getLogger("rex-deploy").setLevel(logging.INFO)
def _quietly(fn) -> None:
"""Call *fn* (a ``close``/``stop`` bound method) swallowing errors — teardown must never raise."""
try:
fn()
except Exception:
pass
class _ManagedRotatingFileHandler(RotatingFileHandler):
"""RotatingFileHandler with managed-mode perms and external-rotation detection.
In managed mode (NixOS) the setgid stateDir needs group-readable files, but
``_open()``/``doRollover()`` honor the umask (0644), so ``chmod 0660`` follows both.
A rotating handler also holds an fd: if the file is rotated externally (logrotate,
``mv``) writes silently go to the old inode, so each emit compares the path's inode
to the open stream's and reopens on mismatch (the ``WatchedFileHandler`` pattern).
"""
def __init__(self, *args, **kwargs):
from hermes_cli.config import is_managed
self._managed = is_managed()
self._unavailable_reported = False
# Set by _ProfileRoutingFileHandler: a vanished logs/ dir means the profile was deleted,
# so skip the write (the router re-routes the record) instead of a FileNotFoundError traceback.
self._skip_write_when_dir_missing = False
self.skipped_missing_dir = False
# Set by _ProfileRoutingFileHandler: once closed (router close, logging.shutdown, a released
# profile) never reopen. The stdlib reopens an append-mode file on emit after close.
self._never_reopen_after_close = False
super().__init__(*args, **kwargs)
self._record_stream_stat()
def _chmod_if_managed(self):
if self._managed:
try:
os.chmod(self.baseFilename, 0o660)
except OSError:
pass
def _record_stream_stat(self, st: Optional[os.stat_result] = None) -> None:
"""Snapshot dev/ino of ``baseFilename`` so emit() can detect external rotation."""
try:
st = st or os.stat(self.baseFilename)
self._stat_dev, self._stat_ino = st.st_dev, st.st_ino
except OSError:
self._stat_dev, self._stat_ino = None, None
def _reopen_stream(self, stat_result=None) -> None:
"""Close and reopen ``baseFilename`` (best-effort).
On failure the stream is left ``None`` so the next emit bails rather than
writing to a stale inode.
"""
if self.stream is not None:
_quietly(self.stream.close)
self.stream = None # type: ignore[assignment]
try:
self.stream = self._open()
except Exception:
return
self._record_stream_stat(stat_result)
def _reopen_if_externally_rotated(self) -> None:
"""Reopen when ``baseFilename`` was renamed, unlinked, or replaced by another inode.
Silent + best-effort: any error falls back to the existing (possibly stale)
stream so logging keeps working instead of dying on a stat failure.
"""
try:
st = os.stat(self.baseFilename)
except FileNotFoundError:
self._reopen_stream() # rotated/unlinked underneath us: recreate at the path
return
except OSError:
return # transient — try again on the next emit
if self._stat_dev is None or self._stat_ino is None:
self._record_stream_stat(st)
elif (st.st_dev, st.st_ino) != (self._stat_dev, self._stat_ino):
self._reopen_stream(st)
def emit(self, record: logging.LogRecord) -> None:
self.skipped_missing_dir = False
# Under this handler's own lock (Handler.handle), which close() also takes: a record that
# picked this handler before its router retired it is dropped, never written to a reopened
# file (on Windows that reopened file keeps the profile's logs/ locked until exit).
if self._never_reopen_after_close and getattr(self, "_closed", False):
return
had_stream = self.stream is not None
# The kernel caches inode metadata, so this stat is sub-microsecond on a hot file.
file_exists = had_stream or os.path.exists(self.baseFilename)
if file_exists:
self._reopen_if_externally_rotated()
# Only when the file is gone (or its reopen just failed), so a healthy emit pays no extra
# stat. ``stream is None`` alone is not a signal: on Windows concurrent-log-handler closes
# the stream after every write.
if (self.stream is None and (had_stream or not file_exists) and self._skip_write_when_dir_missing
and not os.path.isdir(os.path.dirname(self.baseFilename))):
self.skipped_missing_dir = True
return
super().emit(record)
# A record actually reached the file: only now has the destination recovered. Resetting
# in _open() is wrong — open() succeeds on a device whose write/flush still raise EIO,
# which re-armed the report and printed the path once per record.
if self.stream is not None:
self._unavailable_reported = False
def handleError(self, record: logging.LogRecord) -> None:
"""Suppress the known Windows ``concurrent-log-handler`` lock timeout.
CLH's ``emit()`` routes that RuntimeError here, so this is the single point to
silence it before stdlib prints to stderr (which the Desktop slash-worker
captures into chat output). Silencing is not silent: warn once through the
logging system so a wedged lock is visible in the logs instead of a black hole.
"""
exc = sys.exc_info()[1]
if _is_windows_concurrent_log_lock_timeout(exc):
_warn_windows_lock_timeout_once()
return
if _is_unavailable_log_stream(exc):
# The QueueListener must not turn a failing log destination into a traceback for
# every queued record. Name the path once, drop the stale stream; the next emit
# reopens it if the destination has recovered.
if not self._unavailable_reported:
self._unavailable_reported = True
_quietly(lambda: print(
f"hermes_logging: {self.baseFilename} unavailable ({exc}); "
"file logging paused until it recovers", file=_safe_stderr()))
if self.stream is not None:
_quietly(self.stream.close)
self.stream = None # type: ignore[assignment]
return
super().handleError(record)
def _open(self):
stream = super()._open()
self._chmod_if_managed()
return stream
def doRollover(self):
# The stdlib rollover opens a fresh baseFilename owned by whichever process crossed
# maxBytes. With one rotating handler per profile that is usually the long-lived root
# gateway, and a worker on another uid can never reopen its own agent.log (#120151).
# Only root can hand the file back; an unprivileged process never changed the owner.
try:
previous = os.stat(self.baseFilename)
except OSError:
previous = None
super().doRollover()
if previous is not None:
try:
if getattr(os, "geteuid", lambda: -1)() == 0:
os.chown(self.baseFilename, previous.st_uid, previous.st_gid)
os.chmod(self.baseFilename, previous.st_mode & 0o7777)
except OSError:
pass # a log that cannot be chowned is still a working log
self._chmod_if_managed()
# Our own rollover writes a new baseFilename; refresh the snapshot so
# the next emit doesn't mistake it for external rotation.
self._record_stream_stat()
def _new_file_handler(
path: Path, *, level: int, max_bytes: int, backup_count: int, formatter
) -> _ManagedRotatingFileHandler:
"""Create the ``logs/`` directory and a configured ``_ManagedRotatingFileHandler``."""
mkdir_under_hermes_home(path.parent)
if _WINDOWS_CLH_FALLBACK:
# stdlib fallback: no rollover, or the file pins at the size threshold
# and every emit re-triggers the WinError 32 rename failure (#44873).
max_bytes, backup_count = 0, 0
handler = _ManagedRotatingFileHandler(
str(path), maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"
)
handler.setLevel(level)
handler.setFormatter(formatter)
return handler
# A routed profile home is re-checked for an out-of-band delete (missing dir or tombstone) at
# most this often. WHY: the check is two stats per record per router on the listener thread,
# nearly doubling the syscalls of every live-profile emit; a deleted home only needs catching
# within a couple of seconds. Inside that window a tombstoned (not yet removed) home still gets
# its records; an rmtree is caught at once by the write that finds the directory gone.
_PROFILE_LIVENESS_RECHECK_S = 2.0
class _ProfileRoutingFileHandler(logging.Handler):
"""Route queued records to the log file for their Hermes home.
Used only behind the QueueListener, so its small routing lock never blocks an agent
or dashboard event loop. Per-home handlers keep rotation, redaction and managed perms.
"""
def __init__(self, existing: RotatingFileHandler, profile_homes: Sequence[Path]) -> None:
"""Take over *existing*'s path, level, rotation, formatter and filters."""
super().__init__(level=existing.level)
resolved = Path(existing.baseFilename).resolve()
self.baseFilename = str(resolved)
self._hermes_routed_log_path = resolved
self._default_home = resolved.parent.parent.resolve()
self._profile_homes = {Path(home).expanduser().resolve() for home in profile_homes}
self._filename = resolved.name
self._max_bytes = getattr(existing, "maxBytes", 0)
self._backup_count = getattr(existing, "backupCount", 0)
self._profile_handlers: dict[Path, _ManagedRotatingFileHandler] = {}
self._profile_handlers_lock = threading.RLock()
self._router_closed = False # under _profile_handlers_lock; close() is final
# Home -> monotonic time of its last liveness check (see _PROFILE_LIVENESS_RECHECK_S).
self._liveness_checked_at: dict[Path, float] = {}
self.setFormatter(existing.formatter)
for log_filter in existing.filters:
self.addFilter(log_filter)
def _home_for_record(self, record: logging.LogRecord) -> Path:
raw_home = getattr(record, "hermes_home", "")
try:
candidate = Path(raw_home).expanduser().resolve()
except (TypeError, ValueError, OSError):
candidate = self._default_home
if candidate == self._default_home or candidate not in self._profile_homes:
return self._default_home
last = self._liveness_checked_at.get(candidate)
if last is None or _monotonic() - last >= _PROFILE_LIVENESS_RECHECK_S:
# First sighting always checks, so a home deleted before it ever logged is caught on
# its first record; afterwards at most once per interval. An rmtree inside the
# interval is caught by emit() on the record whose write finds the dir gone.
if self._release_if_deleted(candidate):
return self._default_home
return candidate
def _release_if_deleted(self, home: Path) -> bool:
"""Release *home* if it was deleted out of band; True when it was."""
self._liveness_checked_at[home] = _monotonic()
if home.is_dir() and not named_profile_is_deleted(home):
return False
# Deleted out of band (CLI ``hermes profile delete`` while this process runs; it
# tombstones before rmtree). The startup snapshot still names it, so its handler would
# retry the vanished logs/ path on every record (#103777). Release it the way an
# in-process delete does: the stale fd is closed on this routed record and the home is
# dropped from the routing set, so later records fall back to the default home. This
# does not reach into another process: a CLI delete racing a live server's open fd
# still relies on release_profile_log_handlers in-process and #130285.
self.release_profile(home)
return True
def _handler_for_home(self, home: Path) -> Optional[_ManagedRotatingFileHandler]:
"""The open handler for *home*; ``None`` once the router or that home is retired."""
with self._profile_handlers_lock:
handler = self._profile_handlers.get(home)
if handler is not None:
return handler
# Built with no router lock held: ``Handler.__init__`` takes ``logging._lock``, and
# ``logging.shutdown`` (uvicorn's ``dictConfig``) holds that lock while it acquires each
# handler's lock and closes it (``close`` takes ``_profile_handlers_lock``). Holding
# either router lock here deadlocked the QueueListener against ``dictConfig`` (T23).
built = _new_file_handler(
home / "logs" / self._filename, level=self.level, max_bytes=self._max_bytes,
backup_count=self._backup_count, formatter=self.formatter,
)
built._skip_write_when_dir_missing = home != self._default_home
built._never_reopen_after_close = True
with self._profile_handlers_lock:
# Closed (logging.shutdown) or released while this was built: publishing now would leak
# the new handler's fd into a router nothing will close again.
retired = self._router_closed or (home != self._default_home and home not in self._profile_homes)
handler = None if retired else self._profile_handlers.setdefault(home, built)
if handler is not built: # retired, or another thread opened it first
_quietly(built.close)
return handler
def handle(self, record: logging.LogRecord):
"""``Handler.handle`` without this router's own I/O lock: each per-home handler takes its
own lock to write, and routing state has ``_profile_handlers_lock``. Holding ``self.lock``
across ``emit`` (which can open a handler, taking ``logging._lock``) inverted the order
``logging.shutdown`` uses (module lock, then handler lock) (T23)."""
rv = self.filter(record)
if isinstance(rv, logging.LogRecord):
record = rv
if rv:
self.emit(record)
return rv
def emit(self, record: logging.LogRecord) -> None:
try:
home = self._home_for_record(record)
handler = self._handler_for_home(home)
if handler is None and home != self._default_home: # released in flight
home, handler = self._default_home, self._handler_for_home(self._default_home)
if handler is None: # the router is closed
return
if home == self._default_home:
handler.handle(record)
return
# Formatted here, on the listener thread, where the record's profile scope is gone: bind its home so
# RedactingFormatter applies THAT profile's redact_secrets policy and vault values, not the launch's.
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
token = set_hermes_home_override(str(home))
try:
handler.handle(record)
finally:
reset_hermes_home_override(token)
# The write was skipped because logs/ is gone (an rmtree inside the liveness interval,
# see _ManagedRotatingFileHandler.emit): re-check now, and keep the record in the
# default log either way (a live profile whose logs/ was removed by hand included).
if handler.skipped_missing_dir:
self._release_if_deleted(home)
if (fallback := self._handler_for_home(self._default_home)) is not None:
fallback.handle(record)
except Exception:
self.handleError(record)
def close(self) -> None:
with self._profile_handlers_lock:
self._router_closed = True
handlers = list(self._profile_handlers.values())
self._profile_handlers.clear()
for handler in handlers:
_quietly(handler.close)
super().close()
def release_profile(self, home: Path) -> bool:
"""Close and forget the routed files belonging to a deleted profile."""
with self._profile_handlers_lock:
handler = self._profile_handlers.pop(home, None)
self._profile_homes.discard(home)
self._liveness_checked_at.pop(home, None)
if handler is None:
return False
_quietly(handler.close)
return True
# Asynchronous file logging: an ``emit`` can block on the cross-process
# rotation lock (module header); on an asyncio thread that stalls the loop and
# drops WebSocket clients. Every file handler is therefore driven by a single
# QueueListener thread; loggers only do a non-blocking enqueue.
_log_queue: Optional[queue.SimpleQueue] = None
_queue_listener: Optional[QueueListener] = None
_queued_file_handlers: list = []
_queue_atexit_registered = False
# Guards every read-modify-write of the four globals above. setup_logging()
# holds no lock and its _logging_initialized guard runs AFTER handler
# registration, so _register_queued_handler() can race a flush/reset from
# another thread (gateway init vs a plugin/CLI path); without this, two
# threads can interleave stop()/reassign/start() and leave two live listeners.
_queue_state_lock = threading.Lock()
class _NonFormattingQueueHandler(QueueHandler):
"""``QueueHandler`` for an in-process queue.
Stdlib ``prepare()`` formats and strips ``args``/``exc_info`` for cross-process
pickling; ours is in-process, so targets get the unformatted record and apply their
own ``RedactingFormatter`` on the listener thread. A shallow copy is returned because
the emitting thread's synchronous handlers may mutate ``record.message`` meanwhile.
"""
def prepare(self, record: logging.LogRecord) -> logging.LogRecord:
return copy.copy(record)
def _stop_queue_listener() -> None:
"""Flush and stop the background log listener (idempotent; atexit hook, so it takes the lock)."""
global _queue_listener
with _queue_state_lock:
listener, _queue_listener = _queue_listener, None
if listener is not None:
_quietly(listener.stop)
def _start_queue_listener_locked() -> None:
"""(Re)build + start a listener over the current handler set (``_queue_state_lock`` held).
A running listener is stopped first; this only happens while handlers are being
added (queue empty), so ``stop()`` returns immediately.
"""
global _queue_listener
if _queue_listener is not None:
_queue_listener.stop()
_queue_listener = QueueListener(_log_queue, *_queued_file_handlers, respect_handler_level=True)
_queue_listener.start()
def _register_queued_handler(handler: logging.Handler) -> None:
"""Route *handler* through the shared async queue instead of attaching it to root.
Emitting threads never block on file I/O or the rotation lock; the ``QueueListener``
applies each handler's own level and filters on its worker thread.
"""
global _log_queue, _queue_atexit_registered
with _queue_state_lock:
if _log_queue is None:
_log_queue = queue.SimpleQueue()
qh = _NonFormattingQueueHandler(_log_queue)
qh._hermes_queue = True # type: ignore[attr-defined]
# Always on the root logger so records from any logger reach the queue.
logging.getLogger().addHandler(qh)
_queued_file_handlers.append(handler)
_start_queue_listener_locked()
if not _queue_atexit_registered:
# Runs before logging.shutdown (registered earlier at import time),
# so the listener stops before its file handlers are closed.
atexit.register(_stop_queue_listener)
_queue_atexit_registered = True
def flush_log_queue() -> None:
"""Block until all queued records have been written, then resume.
Stops the listener (which processes every pending record before joining) and
restarts it. ``stop()`` joins the worker thread — do NOT call this on a hard-exit
path where the listener may be wedged on the rotation lock; use
``drain_log_queue()`` there, which bounds the wait.
"""
with _queue_state_lock:
listener = _queue_listener
if listener is not None:
listener.stop()
listener.start()
def drain_log_queue(timeout: float = 1.0) -> None:
"""Best-effort, time-bounded drain for hard-exit paths (no restart).
If the listener's worker is wedged on the cross-process rotation lock — the very
failure async logging exists to survive — an unbounded join would re-freeze shutdown.
"""
listener = _queue_listener
if listener is None:
return
t = threading.Thread(target=lambda: _quietly(listener.stop), name="hermes-log-drain", daemon=True)
t.start()
t.join(timeout)
def release_profile_log_handlers(profile_home: str | Path) -> int:
"""Release this process's routed log files below a profile before it is removed.
The Desktop serve process can route records for several profiles through one
``QueueListener``. On Windows, each routed concurrent log handler keeps its
lock file open, so closing only external profile resources still leaves
``logs/.__agent.lock`` and ``logs/.__errors.lock`` unavailable to rmtree.
"""
global _queue_listener
try:
home = Path(profile_home).expanduser().resolve()
except (TypeError, ValueError, OSError):
return 0
with _queue_state_lock:
listener = _queue_listener
if listener is not None:
listener.stop()
_queue_listener = None
released = sum(
handler.release_profile(home)
for handler in _queued_file_handlers
if isinstance(handler, _ProfileRoutingFileHandler)
)
if listener is not None:
_start_queue_listener_locked()
return released
def enable_profile_log_routing(profile_homes: Sequence[str | Path]) -> bool:
"""Make the queued file logs follow a desktop profile context.
``setup_logging`` binds handlers to one process home; the desktop dashboard's
embedded cron ticker may run jobs for every profile, so its static file handlers
are replaced with profile routers once the profile list is known. Returns ``True``
when routing is (or already was) enabled; a single-profile caller is left untouched.
"""
global _queue_listener
homes: list[Path] = []
for entry in profile_homes:
try:
resolved = Path(entry[1] if isinstance(entry, tuple) else entry).expanduser().resolve()
except (TypeError, ValueError, OSError):
continue
if resolved not in homes:
homes.append(resolved)
if len(homes) < 2:
return False
with _queue_state_lock:
if not _queued_file_handlers:
return False
routers = [h for h in _queued_file_handlers if isinstance(h, _ProfileRoutingFileHandler)]
if routers:
for handler in routers:
with handler._profile_handlers_lock:
handler._profile_homes = handler._profile_homes.union(homes)
return True
listener = _queue_listener
if listener is not None:
listener.stop()
_queue_listener = None
replacement = []
for existing in _queued_file_handlers:
if isinstance(existing, RotatingFileHandler):
replacement.append(_ProfileRoutingFileHandler(existing, homes))
_quietly(existing.close)
else:
replacement.append(existing)
_queued_file_handlers[:] = replacement
if listener is not None:
_start_queue_listener_locked()
return True
def _reset_queued_handlers() -> None:
"""Tear down the async logging queue + listener (test-isolation helper)."""
global _log_queue
_stop_queue_listener()
with _queue_state_lock:
root = logging.getLogger()
for h in list(root.handlers):
if getattr(h, "_hermes_queue", False):
root.removeHandler(h)
for h in _queued_file_handlers:
_quietly(h.close)
_queued_file_handlers.clear()
_log_queue = None
def _add_rotating_handler(
path: Path,
*,
level: int,
max_bytes: int,
backup_count: int,
formatter: logging.Formatter,
log_filter: Optional[logging.Filter] = None,
) -> None:
"""Register a queued ``RotatingFileHandler`` for *path*; idempotent per resolved path."""
resolved = path.resolve()
for existing in _queued_file_handlers:
# Already attached directly, or already covered by the profile router — for its default
# home or any profile home it routes (a bare handler beside it would take every record).
if getattr(existing, "_hermes_routed_log_path", None) == resolved or (
isinstance(existing, RotatingFileHandler)
and Path(getattr(existing, "baseFilename", "")).resolve() == resolved
):
return
if isinstance(existing, _ProfileRoutingFileHandler) and existing._filename == resolved.name and (
resolved.parent.parent == existing._default_home or resolved.parent.parent in existing._profile_homes
You can’t perform that action at this time.