diff --git a/agent/replay_cleanup.py b/agent/replay_cleanup.py index 7490df2197..febd7aaea0 100644 --- a/agent/replay_cleanup.py +++ b/agent/replay_cleanup.py @@ -1,11 +1,8 @@ -"""Replay-history sanitization shared across resume code paths. +"""Replay-history sanitization shared by EVERY resume surface (messaging gateway, TUI/WebUI gateway). -A session whose last turn died mid-tool-loop (process killed by a restart command, stale -timeout, interrupt before the tool result was written) persists a dangling -``assistant(tool_calls)`` or interrupted ``assistant→tool`` tail; on resume the model re-issues -the unanswered call → endless "thinking"/reboot loop. These pure helpers strip those tails before -replay, for EVERY resume surface (messaging gateway and TUI/WebUI gateway alike). -""" +A turn that died mid-tool-loop (restart command, stale timeout, interrupt before the result was written) +persists a dangling ``assistant(tool_calls)`` or interrupted ``assistant→tool`` tail; on resume the model +re-issues the unanswered call → endless "thinking"/reboot loop. These pure helpers strip those tails.""" from __future__ import annotations @@ -59,9 +56,9 @@ def _orphan_recovery(name: str, notices: tuple) -> tuple: def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - """Strip interrupted assistant→tool blocks anywhere in replay history (a queued user message may - follow one). Read-only blocks are dropped; blocks with a side-effecting call are KEPT with the - interrupted results rewritten as orphan-recovery notices, since the effect may have happened.""" + """Strip interrupted assistant→tool blocks anywhere in history (a queued user message may follow one). + Read-only blocks are dropped; blocks with a side-effecting call are KEPT with the interrupted results + rewritten as orphan-recovery notices, since the effect may have happened.""" if not agent_history: return agent_history cleaned: List[Dict[str, Any]] = [] @@ -103,10 +100,9 @@ def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Di def strip_dangling_tool_call_tail(agent_history: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - """Strip a trailing ``assistant(tool_calls)`` with NO answers — a call that killed the gateway - itself (``docker restart``) left zero ``tool`` rows, which ``strip_interrupted_tool_tails`` cannot - detect. A partially answered block still resumes. Read-only tails are dropped; side-effecting - ones get synthetic UNKNOWN-effect results.""" + """Strip a trailing ``assistant(tool_calls)`` with NO answers — a call that killed the gateway itself + (``docker restart``) left zero ``tool`` rows, invisible to ``strip_interrupted_tool_tails``. A partially + answered block still resumes. Read-only tails are dropped; side-effecting ones get UNKNOWN-effect results.""" if not agent_history: return agent_history last = agent_history[-1] @@ -126,8 +122,7 @@ def strip_dangling_tool_call_tail(agent_history: List[Dict[str, Any]]) -> List[D def sanitize_replay_history(agent_history: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - """Both replay-tail strippers in canonical order (interrupted blocks, then dangling tail). - Returns the same list object when nothing is stripped.""" + """Both strippers in canonical order (interrupted blocks, then dangling tail); same list object when nothing strips.""" if not agent_history: return agent_history return strip_dangling_tool_call_tail(strip_interrupted_tool_tails(agent_history)) @@ -138,16 +133,15 @@ def sanitize_replay_history(agent_history: List[Dict[str, Any]]) -> List[Dict[st # Short on purpose: a dangerous confirmation must not survive any restart or resume gap. _DANGEROUS_CONFIRMATION_EXPIRY_SECONDS = 60.0 -# Confirmation phrases that unlock destructive host actions; case-insensitive substring match so -# trailing punctuation / extra context still matches. Includes i18n variants from the original incident. +# Phrases that unlock destructive host actions; case-insensitive substring match so trailing punctuation / +# extra context still matches. Includes i18n variants from the original incident. _DANGEROUS_CONFIRMATION_PATTERNS: tuple = ( "confirm forced restart", "confirm forced reboot", "confirm shutdown", "confirm reboot", "confirm power off", "yes, delete everything", "confirm wipe", "confirm factory reset", "確認強制重開機", "確認強制重開", "確認重啟", ) -# Redacting in place (rather than deleting the message) preserves strict user/assistant role -# alternation in the replayed history. +# Redacting in place (not deleting the message) preserves strict user/assistant alternation in the replay. _EXPIRED_CONFIRMATION_SENTINEL = ( "[A high-risk confirmation previously given here has EXPIRED and must " "not be acted on. Ask the user to re-confirm explicitly before " @@ -169,9 +163,9 @@ def strip_stale_dangerous_confirmations( now: float, expiry_seconds: float = _DANGEROUS_CONFIRMATION_EXPIRY_SECONDS, ) -> List[Dict[str, Any]]: - """Redact IN PLACE dangerous-confirmation text older than ``expiry_seconds`` in user messages: a - confirmation surviving a restart can read as a fresh re-confirmation minutes later. Messages - without a timestamp (legacy transcripts, test scaffolding) are left untouched.""" + """Redact IN PLACE dangerous-confirmation text older than ``expiry_seconds`` in user messages: a confirmation + surviving a restart reads as a fresh re-confirmation minutes later. Untimestamped messages (legacy + transcripts, test scaffolding) are left untouched.""" if not agent_history: return agent_history cleaned: List[Dict[str, Any]] = [] @@ -186,8 +180,7 @@ def strip_stale_dangerous_confirmations( ) redacted = dict(msg) redacted["content"] = _EXPIRED_CONFIRMATION_SENTINEL - # The api_content sidecar carries the exact bytes previously sent — the confirmation - # itself; replaying it would undo the redaction. + # The api_content sidecar carries the exact bytes sent — the confirmation itself; replaying it would undo the redaction. drop_stale_api_content(redacted) cleaned.append(redacted) return cleaned diff --git a/agent/session_activity.py b/agent/session_activity.py index 6c756c9ad4..639b21271b 100644 --- a/agent/session_activity.py +++ b/agent/session_activity.py @@ -1,9 +1,6 @@ -"""Shared session activity observation contract. - -Observation-only: timestamp + bounded description/provenance. Notification, timeout, kill and -retry policy live in their own components. Provenance is a small closed enum of *noun* sources; -the default agent clock stamps ``unknown`` unless a writer passes an explicit ``provenance=``. -""" +"""Shared session activity observation contract: timestamp + bounded description/provenance, observation +only (notification, timeout, kill and retry policy live elsewhere). Provenance is a small closed enum of +*noun* sources; the default agent clock stamps ``unknown`` unless a writer passes ``provenance=``.""" from __future__ import annotations @@ -14,10 +11,9 @@ from typing import Any, Mapping, Optional ACTIVITY_DESCRIPTION_MAX = 120 -# Durable SessionDB heartbeat cadence. Contract: MUST stay >= 30s — the SessionDB write path is -# contended and this observation-only projection never justifies extra write pressure. A code -# constant on purpose (no config can turn it into a high-frequency writer); matches the kanban -# auto-heartbeat. force_persist (terminal stamps) is the only bypass. +# Durable SessionDB heartbeat cadence. Contract: MUST stay >= 30s — the SessionDB write path is contended and +# this observation-only projection never justifies extra write pressure. A code constant on purpose (no config +# can make it a high-frequency writer); matches the kanban auto-heartbeat. force_persist (terminal stamps) is the only bypass. SESSION_ACTIVITY_HEARTBEAT_MIN_INTERVAL_SECONDS = 60.0 @@ -49,8 +45,7 @@ def normalize_activity_provenance(provenance: Optional[ActivityProvenance | str] def reset_session_activity_persist_window(agent: Any) -> None: - """Clear the durable persist rate-limit so the next stamp writes through (terminal compression - labels must not stay stuck on mid-compress text).""" + """Clear the persist rate-limit so the next stamp writes through (terminal compression labels must not stick on mid-compress text).""" with suppress(Exception): agent._session_activity_last_persist_mono = 0.0 @@ -68,17 +63,12 @@ def build_activity_snapshot( clock = float(now if now is not None else time.time()) desc = bound_activity_description(last_activity_description) prov = normalize_activity_provenance(last_activity_provenance).value - snap: dict[str, Any] = { + return { "last_activity_at": when, "last_activity_description": desc, "last_activity_provenance": prov, "seconds_since_activity": round(clock - when, 1) if when is not None else None, # Short aliases used by existing gateway/delegate readers. - "last_activity_ts": when, - "last_activity_desc": desc, - "description": desc, - "provenance": prov, + "last_activity_ts": when, "last_activity_desc": desc, "description": desc, "provenance": prov, + **(extra or {}), } - if extra: - snap.update(dict(extra)) - return snap diff --git a/agent/session_persistence.py b/agent/session_persistence.py index 3827ef7380..528c460f12 100644 --- a/agent/session_persistence.py +++ b/agent/session_persistence.py @@ -1,8 +1,6 @@ -"""Durable transcript persistence for ``AIAgent`` (mixin; MRO-resolved from ``run_agent``). - -SQLite session flush with intrinsic ``_DB_PERSISTED_MARKER`` dedup, ephemeral-scaffolding -filtering, the optional JSON session log and trajectory export. -""" +"""Durable transcript persistence for ``AIAgent`` (mixin; MRO-resolved from ``run_agent``): SQLite flush +with intrinsic ``_DB_PERSISTED_MARKER`` dedup, ephemeral-scaffolding filtering, optional JSON session log, +trajectory export.""" import hashlib import json import logging @@ -25,9 +23,7 @@ from agent.trajectory import convert_scratchpad_to_think, save_trajectory as _sa from agent.transcript_repair import sync_flushed_message_markers from utils import atomic_json_write -# Same logger name as the origin module so log records / caplog filters are unchanged. -logger = logging.getLogger("run_agent") - +logger = logging.getLogger("run_agent") # origin module's name: log records / caplog filters unchanged # Flags marking ephemeral recovery scaffolding the loop pops before appending the real response. # Persistence must skip them or a resumed session replays synthetic turns / breaks prefix-cache reuse. @@ -51,16 +47,15 @@ def _is_ephemeral_scaffolding(msg: Any) -> bool: return isinstance(msg, dict) and any(msg.get(flag) for flag in _EPHEMERAL_SCAFFOLDING_FLAGS) -# `_DB_PERSISTED_MARKER` (agent.context_compressor) is the intrinsic "already written to SQLite" marker: -# an id(msg) set can alias a freed dict's address onto a new message, a key on the dict cannot. The `_` -# prefix is mandatory (wire sanitizers strip `_` keys). CONTRACT: the marker asserts the dict's CONTENT -# is durable as written — any in-place mutation that must persist MUST pop it (turn_finalizer, -# context_compressor). +# `_DB_PERSISTED_MARKER` (agent.context_compressor) is the intrinsic "already written to SQLite" marker: an +# id(msg) set can alias a freed dict's address onto a new message, a key on the dict cannot. The `_` prefix is +# mandatory (wire sanitizers strip `_` keys). CONTRACT: the marker asserts the dict's CONTENT is durable as +# written — any in-place mutation that must persist MUST pop it (turn_finalizer, context_compressor). def _safe_session_filename_component(session_id: str) -> str: - """Path-safe filename component for a (possibly untrusted ``X-Hermes-Session-Id``) session ID: - non ``[A-Za-z0-9_-]`` → ``_``, capped, plus a content hash when changed so distinct IDs cannot collide.""" + """Path-safe component for a (possibly untrusted ``X-Hermes-Session-Id``) ID: non ``[A-Za-z0-9_-]`` → ``_``, + capped, plus a content hash when changed so distinct IDs cannot collide.""" raw = str(session_id or "").strip() sanitized = re.sub(r"[^\w-]", "_", raw).strip("._")[:96] or "session" if raw and sanitized == raw: @@ -69,9 +64,9 @@ def _safe_session_filename_component(session_id: str) -> str: def _override_replaces_content(msg: Dict, content: Any, override: Any) -> bool: - """Whether the persist user-message override may replace ``content``: a plain-text override must - not replace native image/audio blocks (a list override is the clean multimodal payload and does), - and never a message MERGED with a compaction summary (overwriting would drop the summary).""" + """May the persist override replace ``content``? A plain-text override must not replace native image/audio + blocks (a list override is the clean multimodal payload and does), nor a message MERGED with a compaction + summary (overwriting would drop the summary).""" return ( override is not None and not msg.get(COMPRESSED_SUMMARY_METADATA_KEY) @@ -80,8 +75,8 @@ def _override_replaces_content(msg: Dict, content: Any, override: Any) -> bool: def _summary_display_kind(msg: Dict) -> Any: - """Standalone handoffs are hidden so they never occupy the active user slot in retry/undo - dispatch; merge-into-tail carriers keep their prior visibility.""" + """Standalone handoffs are hidden so they never occupy the active user slot in retry/undo dispatch; + merge-into-tail carriers keep their prior visibility.""" if ( msg.get(COMPRESSED_SUMMARY_METADATA_KEY) and user_originated_turn_view(msg) is None @@ -123,10 +118,9 @@ def _persist_lock(agent): # --- flush phases (module-level so the flush also works bound onto duck-typed agents) --- - def _db_flush_seed_ids(agent) -> set: - """One-shot ``_flushed_db_message_ids`` seed (same session, after a non-empty flush); the scan - translates it to markers and the flush clears it.""" + """One-shot ``_flushed_db_message_ids`` seed (same session, after a non-empty flush); the scan translates + it to markers and the flush clears it.""" current_session_id = getattr(agent, "session_id", None) seed_ids = None if getattr(agent, "_flushed_db_message_session_id", None) == current_session_id and agent._last_flushed_db_idx != 0: @@ -149,15 +143,13 @@ def _db_flush_row(agent, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any """Build the session-db row for ``msg``, applying the persist override to THIS row only.""" role = msg.get("role", "unknown") content = msg.get("content") - # api_content sidecar: exact bytes sent to the API when they differ from clean content, so - # replay reproduces the sent prefix byte-for-byte. + # api_content sidecar: exact bytes sent to the API when they differ from clean content (replay parity). api_content = msg.get("api_content") if isinstance(msg.get("api_content"), str) else None timestamp = msg.get("timestamp") if is_current_turn_user and msg.get("role") == "user": override = getattr(agent, "_persist_user_message_override", None) if _override_replaces_content(msg, content, override): - # Live content is what the wire sent, the override is the clean transcript; keep the - # sent bytes in api_content so replay matches the wire. + # Live content is what the wire sent, the override is the clean transcript; keep the sent bytes. if api_content is None and isinstance(content, str) and content != override: api_content = content content = override @@ -166,8 +158,8 @@ def _db_flush_row(agent, msg: Dict, is_current_turn_user: bool) -> Dict[str, Any timestamp = ov_timestamp if api_content == content: api_content = None - # get_messages_as_conversation replays rows through sanitize_context().strip(); capture the - # sent bytes when they would differ (compared in wire form). + # get_messages_as_conversation replays rows through sanitize_context().strip(); capture the sent bytes + # when they would differ (compared in wire form). if ( api_content is None and role in ("user", "assistant") and isinstance(content, str) and content and sanitize_context(content).strip() != content.strip() @@ -200,15 +192,15 @@ def _db_flush_collect(agent, messages: List[Dict], conversation_history: Optiona seed_ids = _db_flush_seed_ids(agent) history_ids = {id(item) for item in (conversation_history or []) if isinstance(item, dict)} ov_idx = getattr(agent, "_persist_user_message_idx", None) - # Also match the staged CLI dict by identity — the close safety-net may flush a shortened - # snapshot whose turn index refers to the full history. + # Also match the staged CLI dict by identity — the close safety-net may flush a shortened snapshot whose + # turn index refers to the full history. pending_cli_message = getattr(agent, "_pending_cli_user_message", None) batch_rows: List[Dict[str, Any]] = [] batch_msgs: List[Dict] = [] for msg_idx in range(_db_flush_scan_start(agent, messages), len(messages)): msg = messages[msg_idx] - # The flush is append-only: a mid-turn persist of scaffolding could commit a synthetic - # turn the end-of-turn drop cannot un-write. Skip regardless of position. + # Append-only flush: a mid-turn persist of scaffolding would commit a synthetic turn the end-of-turn + # drop cannot un-write. Skip regardless of position. if not isinstance(msg, dict) or _is_ephemeral_scaffolding(msg) or msg.get(_DB_PERSISTED_MARKER): continue # Already durable (history copy or caller-seeded): stamp so future flushes skip it. @@ -235,8 +227,8 @@ def _db_flush_write(agent, batch_rows: List[Dict[str, Any]], batch_msgs: List[Di def _db_flush_adopt_compression_tip(agent) -> bool: - """Adopt the live continuation of a session closed by compression. Same-id tip = no continuation; - a tip whose row is missing or already ended is not adopted either.""" + """Adopt the live continuation of a compression-closed session. Same-id tip = no continuation; a tip + whose row is missing or already ended is not adopted either.""" old_id = agent.session_id try: tip = agent._session_db.get_compression_tip(old_id) @@ -261,10 +253,9 @@ def _db_flush_adopt_compression_tip(agent) -> bool: def _db_flush_failed(agent, e: Exception, batch_rows: List[Dict[str, Any]], adoption_budget: int) -> bool: """Classify a failed flush; True when the caller should retry once on an adopted compression tip.""" - # Force a full re-scan next flush: an exception mid-loop leaves mixed dispositions. - agent._db_flush_scan_prefix = None - # The only place the SQLite error is visible before it becomes a bare False — classify it so - # the turn-end explanation can distinguish lock contention from disk-full/read-only. + agent._db_flush_scan_prefix = None # full re-scan next flush: an exception mid-loop leaves mixed dispositions + # The only place the SQLite error is visible before it becomes a bare False — classify it so the turn-end + # explanation can distinguish lock contention from disk-full/read-only. from hermes_state import ( CompressionSessionClosedError, StateDbCorruptError, @@ -284,19 +275,17 @@ def _db_flush_failed(agent, e: Exception, batch_rows: List[Dict[str, Any]], adop agent._last_persistence_error_cause, getattr(agent, "session_id", None), exc_info=True, ) if isinstance(e, CompressionSessionClosedError): - # Compression race: another path rotated this session mid-write. Retry exactly once on the - # live tip; a second closed-parent write fails closed. + # Compression race: another path rotated this session mid-write. Retry exactly once on the live tip; a + # second closed-parent write fails closed. if adoption_budget > 0 and _db_flush_adopt_compression_tip(agent): return True - # The flag lets the turn explanation name compression rotation instead of misleading - # full-disk advice. - agent._compression_adoption_failed = True + agent._compression_adoption_failed = True # lets the turn explanation name rotation, not full-disk advice logger.warning("Session DB append_message failed: %s", e) return False def _session_log_entry(agent, msg: Dict[str, Any]) -> Dict[str, Any]: - """Copy of ``msg`` with scratchpad tags normalised and credentials redacted (respects HERMES_REDACT_SECRETS).""" + """Copy of ``msg`` with scratchpad tags normalised and credentials redacted (honours HERMES_REDACT_SECRETS).""" if "content" not in msg: return msg content = msg["content"] @@ -306,8 +295,8 @@ def _session_log_entry(agent, msg: Dict[str, Any]) -> Dict[str, Any]: def _existing_log_is_larger(log_file, count: int) -> bool: - """Never overwrite a larger log with fewer messages (resumed agent with partial history); - a corrupted existing file allows the overwrite.""" + """Never overwrite a larger log with fewer messages (resumed agent with partial history); a corrupted + existing file allows the overwrite.""" if not log_file.exists(): return False try: @@ -325,8 +314,8 @@ class SessionPersistenceMixin: """Session DB flush, session log and trajectory persistence (see module docstring).""" def _apply_persist_user_message_override(self, messages: List[Dict]) -> None: - """Rewrite the current-turn user message in place: some paths send an API-only variant that - must not leak into transcripts or resumed history.""" + """Rewrite the current-turn user message in place: some paths send an API-only variant that must not + leak into transcripts or resumed history.""" idx = getattr(self, "_persist_user_message_idx", None) override = getattr(self, "_persist_user_message_override", None) timestamp = getattr(self, "_persist_user_message_timestamp", None) @@ -345,10 +334,9 @@ class SessionPersistenceMixin: msg["platform_message_id"] = platform_id def _persist_session(self, messages: List[Dict], conversation_history: List[Dict] = None): - """Save session state to both JSON log and SQLite on any exit path. Trailing empty-response - scaffolding is dropped from the live list; the persist override is applied to the DB row only.""" + """Save to JSON log and SQLite on any exit path. Trailing empty-response scaffolding is dropped from + the live list; the persist override is applied to the DB row only.""" from agent.agent_runtime_helpers import note_turn_persisted - with _persist_lock(self): self._drop_trailing_empty_response_scaffolding(messages) self._session_messages = messages @@ -360,9 +348,9 @@ class SessionPersistenceMixin: note_turn_persisted(self) def _drop_trailing_empty_response_scaffolding(self, messages: List[Dict]) -> None: - """Remove empty-response retry scaffolding from the tail, then (only if any was present) rewind - the tool-result / assistant(tool_calls) pair the failed iteration left hanging — otherwise the - next user turn lands as ``...tool, user`` and providers return empty content forever.""" + """Pop empty-response retry scaffolding from the tail, then (only if any was present) rewind the + tool-result / assistant(tool_calls) pair the failed iteration left hanging — otherwise the next user + turn lands as ``...tool, user`` and providers return empty content forever.""" def tail(*keys: str) -> bool: return bool(messages) and isinstance(messages[-1], dict) and any(messages[-1].get(k) for k in keys) @@ -391,26 +379,24 @@ class SessionPersistenceMixin: def _flush_messages_to_session_db_unlocked( self, messages: List[Dict], conversation_history: Optional[List[Dict]] = None, _adoption_budget: int = 1, ): - """Persist un-flushed messages to SQLite. Dedup is the intrinsic ``_DB_PERSISTED_MARKER`` on - each written dict — not positional slices (drift after sequence repair) nor an ``id(msg)`` set - (address reuse). The persist override touches the written row only. A compression-closed - session adopts its live tip and retries exactly once.""" - # Persistence-isolated agents (background review fork) share the parent's session_id for cache - # warmth; a write here would land the curator's turn in the user's real history. + """Persist un-flushed messages to SQLite. Dedup is the intrinsic ``_DB_PERSISTED_MARKER`` on each written + dict — not positional slices (drift after sequence repair) nor an ``id(msg)`` set (address reuse). The + persist override touches the written row only. A compression-closed session adopts its live tip and + retries exactly once.""" + # Persistence-isolated agents (background review fork) share the parent's session_id for cache warmth; + # a write here would land the curator's turn in the user's real history. if getattr(self, "_persist_disabled", False) or not self._session_db: return None batch_rows: List[Dict[str, Any]] = [] try: - # Retry row creation if the earlier attempt failed transiently. - if not self._session_db_created: + if not self._session_db_created: # retry row creation if the earlier attempt failed transiently self._ensure_db_session() batch_rows, batch_msgs = _db_flush_collect(self, messages, conversation_history) _db_flush_write(self, batch_rows, batch_msgs) # Markers are now the sole truth; reset the one-shot seed so no id() outlives this flush. self._flushed_db_message_ids = set() self._last_flushed_db_idx = len(messages) - # Snapshot for the bounded scan — only on full success, so a partially-processed list can - # never be treated as settled. + # Snapshot for the bounded scan — only on full success, so a partial list is never treated as settled. self._db_flush_scan_prefix = messages[:] return True except Exception as e: @@ -462,15 +448,14 @@ class SessionPersistenceMixin: ] def _save_session_log(self, messages: List[Dict[str, Any]] = None): - """Optional per-session JSON snapshot (``sessions.write_json_snapshots``, default False) for - external tooling; state.db is canonical. Rewrites the full list after every persistence point.""" + """Optional per-session JSON snapshot (``sessions.write_json_snapshots``, default False) for external + tooling; state.db is canonical. Rewrites the full list after every persistence point.""" if not getattr(self, "_session_json_enabled", False): return messages = messages or self._session_messages if not messages: return - # Re-derive the path each call so /branch and /compress land in the right file. - try: + try: # re-derive the path each call so /branch and /compress land in the right file log_file = self.logs_dir / f"session_{_safe_session_filename_component(self.session_id)}.json" except Exception: return diff --git a/agent/shell_hooks.py b/agent/shell_hooks.py index 5fcc020830..5e3244b751 100644 --- a/agent/shell_hooks.py +++ b/agent/shell_hooks.py @@ -1,12 +1,8 @@ -"""Shell-script hooks bridge. - -Reads the ``hooks:`` block from config, prompts for first-use consent per ``(event, command)`` -pair, and registers callbacks on the plugin hook manager so every ``invoke_hook()`` site dispatches -to the configured scripts. Wire protocol: stdin JSON ``{hook_event_name, tool_name, tool_input, -session_id, cwd, extra}``; optional stdout JSON ``{"decision"|"action": "block"|"modify", ...}`` / -``{"context": ...}`` translated by ``_parse_response``. Exit code 2 blocks a ``pre_tool_call`` even -without JSON (Claude-Code / Cursor compatible). Hooks fail open unless ``fail_closed``. -""" +"""Shell-script hooks bridge: ``hooks:`` config → first-use consent per ``(event, command)`` → +callbacks on the plugin hook manager, so every ``invoke_hook()`` site dispatches to the scripts. +Wire: stdin JSON ``{hook_event_name, tool_name, tool_input, session_id, cwd, extra}``; optional stdout +JSON ``{"decision"|"action": "block"|"modify", ...}`` / ``{"context": ...}`` via ``_parse_response``. +Exit code 2 blocks a ``pre_tool_call`` even without JSON (Claude-Code / Cursor). Fail open unless ``fail_closed``.""" from __future__ import annotations @@ -53,9 +49,8 @@ _TRUTHY = {"1", "true", "yes", "on"} # kwargs promoted to top-level payload keys; everything else lands under ``extra``. _TOP_LEVEL_PAYLOAD_KEYS = {"tool_name", "args", "session_id", "parent_session_id"} -# (home, event, matcher, command) wired to the plugin manager in this process. Matcher is in the key -# (one script may register per-tool under one event); home so multiplexed-gateway profiles can register -# identical triples. +# (home, event, matcher, command) wired in this process: matcher in the key (one script may register +# per-tool under one event), home so multiplexed-gateway profiles can register identical triples. _registered: Set[Tuple[str, str, Optional[str], str]] = set() _registered_lock = threading.Lock() # Non-POSIX fallback for allowlist read-modify-write. Must be separate from _registered_lock, which @@ -122,8 +117,8 @@ class _ToolMatcherMixin: return True if tool_name is None: return False - if self.compiled_matcher is None: - return tool_name == self.matcher # regex failed to compile: literal fallback + if self.compiled_matcher is None: # regex failed to compile: literal fallback + return tool_name == self.matcher return self.compiled_matcher.fullmatch(tool_name) is not None @@ -142,14 +137,13 @@ class ShellHookSpec(_ToolMatcherMixin): # --- Public API --- def register_from_config(cfg: Optional[Dict[str, Any]], *, accept_hooks: bool = False) -> List[ShellHookSpec]: - """Register every configured shell hook on the plugin manager; idempotent. Returns the specs - newly wired up; skipped entries (unknown, malformed, not allowlisted, registered) are logged only.""" + """Register every configured shell hook (idempotent); returns the newly wired specs. Skipped + entries (unknown, malformed, not allowlisted, already registered) are logged only.""" if not isinstance(cfg, dict): return [] - # Safe mode: hooks are user customizations too — fire zero user-configured code. from utils import env_var_enabled - if env_var_enabled("HERMES_SAFE_MODE"): + if env_var_enabled("HERMES_SAFE_MODE"): # hooks are user customizations too — fire zero user-configured code logger.info("HERMES_SAFE_MODE=1 — shell-hook registration skipped") return [] effective_accept = _resolve_effective_accept(cfg, accept_hooks) @@ -158,11 +152,9 @@ def register_from_config(cfg: Optional[Dict[str, Any]], *, accept_hooks: bool = return [] registered: List[ShellHookSpec] = [] from hermes_cli.plugins import get_plugin_manager # lazy: avoids import cycle - manager = get_plugin_manager() home_key = _home_key() - # Idempotence + allowlist read happen under the lock; the TTY prompt runs outside it; mutation - # re-takes the lock and re-checks in case two callers raced. + # Idempotence + allowlist read under the lock; TTY prompt outside it; mutation re-takes the lock and re-checks. for spec in specs: key = (home_key, spec.event, spec.matcher, spec.command) with _registered_lock: @@ -195,16 +187,15 @@ def iter_configured_hooks(cfg: Optional[Dict[str, Any]]) -> List[ShellHookSpec]: def re_register_config_hooks() -> None: - """Re-register config hooks after a plugin force-reload cleared the manager's hooks. Only the - current home's keys are cleared (profile A's reload never drops profile B); never re-prompts.""" + """Re-register after a plugin force-reload cleared the manager's hooks; only this home's keys + are cleared (profile A's reload never drops B), never re-prompts.""" _forget_home_registrations(_registered, _registered_lock) from hermes_cli.config import load_config - register_from_config(load_config()) def reset_for_tests() -> None: - """Clear the idempotence set. Test-only helper.""" + """Test-only: clear the idempotence set.""" with _registered_lock: _registered.clear() @@ -300,8 +291,7 @@ _POPEN_ERRORS = ((FileNotFoundError, "command not found"), (PermissionError, "co def _spawn(spec: ShellHookSpec, stdin_json: str) -> Dict[str, Any]: - """Run ``spec.command`` with ``stdin_json`` on stdin; the single subprocess site. - Returns a diagnostic dict with the same keys for every outcome.""" + """The single subprocess site: run ``spec.command`` with ``stdin_json`` on stdin. Same result keys for every outcome.""" result: Dict[str, Any] = { "returncode": None, "stdout": "", "stderr": "", "timed_out": False, "elapsed_seconds": 0.0, "error": None, } @@ -317,9 +307,8 @@ def _spawn(spec: ShellHookSpec, stdin_json: str) -> Dict[str, Any]: if not argv: return failed("empty command") t0 = time.monotonic() - # Own process group on POSIX so a timed-out hook's descendants are reaped with it; Windows tree - # cleanup goes through kill_process_tree (taskkill /T). Hooks that finish in time keep detached - # helpers alive. + # Own process group on POSIX so a timed-out hook's descendants are reaped with it (Windows: kill_process_tree + # / taskkill /T). Hooks that finish in time keep detached helpers alive. popen_kwargs: Dict[str, Any] = {"creationflags": windows_hide_flags()} if IS_WINDOWS else {"process_group": 0} try: proc = subprocess.Popen( @@ -363,10 +352,9 @@ def _fail_closed_block(spec: ShellHookSpec, reason: str) -> Dict[str, Any]: def _evaluate_result(spec: ShellHookSpec, r: Dict[str, Any]) -> Optional[Dict[str, Any]]: - """``_spawn`` diagnostic dict → the hook's contribution (shared by the live callback and - ``run_once``). Spawn error/timeout fail open unless fail_closed; exit 2 on a blocking event blocks - (message from stdout JSON, then stderr, then default); other non-zero exits warn then parse - stdout; unparseable stdout on a fail_closed hook blocks.""" + """``_spawn`` result → hook contribution (live callback and ``run_once``). Spawn error/timeout fail + open unless fail_closed; exit 2 on a blocking event blocks (message: stdout JSON, then stderr, then + default); other non-zero exits warn then parse stdout; unparseable stdout on a fail_closed hook blocks.""" blocking_event = spec.event in _BLOCKING_EVENTS fail_closed = spec.fail_closed and blocking_event if r["error"]: @@ -417,9 +405,8 @@ def _block_message(primary: Any, secondary: Any) -> str: return raw if isinstance(raw, str) and raw else _DEFAULT_BLOCK_MESSAGE -# pre_tool_call response dialects, checked in order: Hermes ``action`` then Claude-Code ``decision``. -# (verb key, block-message primary, block-message secondary, modify payload key) — the Claude-Code shape -# is translated to the canonical one expected by get_pre_tool_call_block_message. +# pre_tool_call dialects in check order — Hermes ``action`` then Claude-Code ``decision`` — as (verb key, +# block-message primary, secondary, modify payload key); both translate to the canonical Hermes shape. _PRE_TOOL_DIALECTS = (("action", "message", "reason", "args"), ("decision", "reason", "message", "tool_input")) @@ -487,7 +474,7 @@ def load_allowlist() -> Dict[str, Any]: def save_allowlist(data: Dict[str, Any]) -> None: - """Atomically persist the allowlist; on OSError log and keep the in-process approval.""" + """Atomic write; on OSError log and keep the in-process approval.""" p = allowlist_path() try: p.parent.mkdir(parents=True, exist_ok=True) @@ -574,8 +561,7 @@ def _record_approval(event: str, command: str) -> None: def revoke(command: str) -> int: - """Remove every allowlist entry matching ``command``; returns the count removed. - Live callbacks in the current process stay registered until restart.""" + """Remove every allowlist entry for ``command``; returns the count. Live callbacks stay registered until restart.""" with _locked_update_approvals() as data: before = len(data.get("approvals", [])) data["approvals"] = [e for e in data.get("approvals", []) if not _entry_matches(e, None, command)] @@ -588,8 +574,7 @@ _SCRIPT_EXTENSIONS: Tuple[str, ...] = ( def _command_script_path(command: str) -> str: - """Script path from ``command``: first token with a script extension, then a path-like token, - then the first token (``python3 /p/hook.py``, ``/usr/bin/env bash x.sh``).""" + """First token with a script extension, else the first path-like token, else the first token.""" try: parts = split_command_line(command) or [command] except ValueError: @@ -631,8 +616,7 @@ def script_mtime_iso(command: str) -> Optional[str]: def script_is_executable(command: str) -> bool: - """True iff ``command`` is runnable as configured: a bare script needs X_OK, an - interpreter-prefixed one only R_OK (mirrors what ``_spawn`` does).""" + """Runnable as configured: a bare script needs X_OK, an interpreter-prefixed one only R_OK (as ``_spawn`` does).""" path = _command_script_path(command) expanded = os.path.expanduser(path) if not path or not os.path.isfile(expanded): @@ -646,8 +630,7 @@ def script_is_executable(command: str) -> bool: def run_once(spec: ShellHookSpec, kwargs: Dict[str, Any]) -> Dict[str, Any]: - """Fire one hook with a synthetic payload (``hermes hooks test`` / doctor); routes through - ``_serialize_payload`` and ``_evaluate_result`` so the result matches production exactly.""" + """Fire one hook with a synthetic payload (``hermes hooks test`` / doctor) through the production path.""" result = _spawn(spec, _serialize_payload(spec.event, kwargs)) result["parsed"] = _evaluate_result(spec, result) return result diff --git a/agent/subagent_lifecycle.py b/agent/subagent_lifecycle.py index e4457a1c38..6a103bb1da 100644 --- a/agent/subagent_lifecycle.py +++ b/agent/subagent_lifecycle.py @@ -1,8 +1,5 @@ -"""Public, plugin-safe lifecycle API for delegated Hermes subagents. - -Exposes immutable contracts, not ``AIAgent`` objects; plugins obtain it via -``PluginContext.subagent_lifecycle``. -""" +"""Public, plugin-safe lifecycle API for delegated Hermes subagents: immutable contracts, not ``AIAgent`` +objects. Plugins obtain it via ``PluginContext.subagent_lifecycle``.""" from __future__ import annotations @@ -156,9 +153,7 @@ class _Registry: _REGISTRY = _Registry() -# Daemon pool: a wedged/abandoned child must never block interpreter exit. -from tools.daemon_pool import DaemonThreadPoolExecutor as _DaemonExecutor - +from tools.daemon_pool import DaemonThreadPoolExecutor as _DaemonExecutor # daemon: a wedged child never blocks exit _EXECUTOR = _DaemonExecutor(max_workers=8, thread_name_prefix="hermes-lifecycle") _SECRET = secrets.token_bytes(32) _ACTIVE_PARENT_AGENT: contextvars.ContextVar[Any] = contextvars.ContextVar("hermes_subagent_lifecycle_parent", default=None) @@ -209,8 +204,8 @@ _HANDLE_FIELD_CHECKS: tuple[tuple[str, Callable[[Any], bool]], ...] = ( ("capability", lambda v: isinstance(v, str)), ) -# Launch-request rejections in check order: (predicate, error). The type check leads so later -# predicates may dereference request fields. +# Launch-request rejections in check order: (predicate, error). The type check leads so later predicates may +# dereference request fields. _REQUEST_REJECTIONS: tuple[tuple[Callable[[Any], bool], str], ...] = ( (lambda r: not isinstance(r, SubagentLaunchRequest) or not isinstance(r.goal, str) or not r.goal.strip() or len(r.goal) > _MAX_GOAL_CHARS, "goal must be a non-empty string of at most 16000 characters."), @@ -230,9 +225,9 @@ def _handle_is_well_formed(handle: Any) -> bool: class SubagentLifecycleService: - """Stable public service returned by :attr:`PluginContext.subagent_lifecycle`. Children run - in-process only; completed results stay until process exit, and ``reconnect`` reports that a - serialized handle cannot reconnect after a restart instead of launching work again.""" + """Stable public service behind :attr:`PluginContext.subagent_lifecycle`. Children run in-process only; + completed results stay until process exit; ``reconnect`` reports that a serialized handle cannot + reconnect after a restart instead of launching work again.""" def __init__(self, parent_agent_resolver: Callable[[], Any]) -> None: self._parent_agent_resolver = parent_agent_resolver @@ -250,9 +245,8 @@ class SubagentLifecycleService: self._cleanup_locked() if request.correlation_id and correlation_key in _REGISTRY.correlations: raise SubagentLifecycleError("Duplicate correlation_id for this parent session.") - # Delegate construction stays internal: plugins never import private delegation helpers. + # Lazy: delegate construction stays internal, plugins never import private delegation helpers. from tools.delegate_tool import _build_child_preserving_parent_tools, DEFAULT_MAX_ITERATIONS - child = _build_child_preserving_parent_tools( task_index=0, goal=request.goal, context=request.context, toolsets=list(request.allowed_toolsets) if request.allowed_toolsets else None, @@ -362,7 +356,6 @@ class SubagentLifecycleService: record.started_at = record.updated_at = time.time() try: from tools.delegate_tool import _run_child_lifecycle - raw = _run_child_lifecycle(0, goal, record.agent, parent) is_dict = isinstance(raw, dict) raw = raw if is_dict else {} @@ -411,7 +404,6 @@ class SubagentLifecycleService: raise SubagentLifecycleError("metadata exceeds 8192 bytes.") if request.allowed_toolsets: from toolsets import TOOLSETS - unknown = set(request.allowed_toolsets) - set(TOOLSETS) if unknown: raise SubagentLifecycleError(f"Unknown toolsets: {', '.join(sorted(unknown))}.") diff --git a/agent/title_generator.py b/agent/title_generator.py index 618e2d5867..f4d4f66a95 100644 --- a/agent/title_generator.py +++ b/agent/title_generator.py @@ -1,11 +1,9 @@ """Auto-generate short session titles from the user's opening message. -Two stages, both off the critical path: an **instant** deterministic title derived from the first -user message (written before the model is called, cannot fail), then an **upgrade** from one -small-model call (cheap tier, thinking off, JSON-constrained). Provenance ``derived < llm < user`` -is enforced by the storage layer, so stage 2 only replaces stage 1 and neither replaces a name the -user typed. -""" +Two stages, both off the critical path: an **instant** deterministic title (written before the model +is called, cannot fail), then an **upgrade** from one small-model call (cheap tier, thinking off, +JSON-constrained). Storage enforces provenance ``derived < llm < user``: stage 2 only replaces stage 1 +and neither replaces a name the user typed.""" import json import logging @@ -88,9 +86,8 @@ _MACHINE_PREFIXES = ( def _title_config() -> dict: - """``auxiliary.title_generation`` from config (lazy read-only import: no hermes_cli cycle, no migration writes).""" + """``auxiliary.title_generation`` (lazy read-only import: no hermes_cli cycle, no migration writes).""" from hermes_cli.config import load_config_readonly - return ((load_config_readonly() or {}).get("auxiliary") or {}).get("title_generation") or {} @@ -105,7 +102,6 @@ def _title_language() -> str: def _auto_title_enabled() -> bool: try: from utils import is_truthy_value - return is_truthy_value(_title_config().get("enabled"), default=True) except Exception: logger.debug("Failed to read title_generation.enabled", exc_info=True) @@ -115,8 +111,7 @@ def _auto_title_enabled() -> bool: def strip_control_wrappers(text: str) -> str: """Remove leading control wrappers (nested too) so a slash-command turn reduces to the prose the user typed.""" current = (text or "").strip() - # Bounded: each pass must remove at least one wrapper or we stop. - for _ in range(len(_CONTROL_WRAPPERS) * 2): + for _ in range(len(_CONTROL_WRAPPERS) * 2): # bounded: each pass must remove a wrapper or we stop stripped = _strip_one_wrapper(current) if stripped == current: break @@ -138,13 +133,12 @@ def _strip_one_wrapper(text: str) -> str: def _summarize_user_message(user_message: str) -> str: - """Text worth titling: a ``/skill`` invocation embeds the whole skill body, so describe it first, then strip wrappers.""" + """Text worth titling: describe a ``/skill`` invocation (it embeds the whole skill body), then strip wrappers.""" if not user_message: return "" described = None try: from agent.skill_commands import describe_skill_invocation - described = describe_skill_invocation(user_message) except Exception: logger.debug("Skill-scaffolding summary failed; titling raw", exc_info=True) @@ -177,8 +171,7 @@ def _first_line(text: str) -> str: def _extract_title_text(content: str) -> str: - """Title from a model response: strict JSON, then a loose JSON scan, then first-line prose - (a provider ignoring ``response_format`` still titles).""" + """Strict JSON, then a loose JSON scan, then first-line prose (a provider ignoring ``response_format`` still titles).""" if not content: return "" raw = content.strip() @@ -200,7 +193,6 @@ def _extract_title_text(content: str) -> str: # Prose fallback: scrub blocks so reasoning can't leak into a title. try: from agent.agent_runtime_helpers import strip_think_blocks - raw = strip_think_blocks(None, raw).strip() except Exception: logger.debug("strip_think_blocks unavailable for title output", exc_info=True) @@ -240,8 +232,8 @@ def generate_title( main_runtime: dict = None, runtime_validator: Optional[RuntimeValidator] = None, ) -> Optional[str]: - """Title from the user's opening message alone (waiting for the assistant made this slow and - bought nothing). ``runtime_validator`` runs right before the request; False skips silently.""" + """Title from the opening message alone (waiting for the assistant made this slow and bought + nothing). ``runtime_validator`` runs right before the request; False skips silently.""" if not _auto_title_enabled(): logger.debug("Auto-title skipped: auxiliary.title_generation.enabled=false") return None @@ -250,8 +242,7 @@ def generate_title( if not runtime_validator(): logger.debug("Title generation skipped: runtime validator returned False") return None - except Exception: - # Fail open: a broken validator must not disable titling. + except Exception: # fail open: a broken validator must not disable titling logger.debug("Title runtime validator raised; proceeding", exc_info=True) user_snippet = _summarize_user_message(user_message)[:MAX_TITLE_INPUT_CHARS] if not user_snippet.strip(): @@ -294,14 +285,11 @@ def _has_upgraded_title(session_db, session_id: str) -> bool: def _persist_session_title(session_db, session_id, title, *, source, dedupe=True): - """Persist a title at *source* authority via ``set_auto_title`` (precedence check + write in - one transaction, so a manual ``/title`` is never overwritten); None when a higher-authority - title held the row. - - ``ValueError`` = unique-title index rejected the name → append ``#N`` via - ``get_next_title_in_lineage``. ``dedupe=False`` re-raises instead: the derived title is on the - critical path, collides constantly ("hi"), and the model replaces it a second later anyway. - """ + """Persist at *source* authority via ``set_auto_title`` (precedence check + write in one + transaction, so a manual ``/title`` is never overwritten); None when a higher authority held the row. + ``ValueError`` = unique-title index collision → append ``#N`` via ``get_next_title_in_lineage``; + ``dedupe=False`` re-raises instead (the derived title is on the critical path, collides constantly + on "hi", and the model replaces it a second later anyway).""" auto_fn = getattr(session_db, "set_auto_title", None) def _set(candidate): @@ -330,8 +318,7 @@ def _persist_session_title(session_db, session_id, title, *, source, dedupe=True def apply_instant_title(session_db, session_id: str, user_message: str, title_callback: Optional[TitleCallback] = None) -> Optional[str]: - """Write the derived title synchronously (cheap enough to run inline). Returns the title - written, or None (no usable text, or a title of at least ``derived`` authority exists). Never raises.""" + """Write the derived title inline. Returns it, or None (no usable text, or a ``derived``+ title exists). Never raises.""" if not session_db or not session_id: return None try: @@ -356,20 +343,17 @@ def auto_title_session( title_callback: Optional[TitleCallback] = None, runtime_validator: Optional[RuntimeValidator] = None, ) -> None: - """Generate and store the model title (daemon-thread target); skips sessions already carrying - an ``llm``/``user`` title. Never lets an exception escape (the threading excepthook would spray - a traceback into the terminal); the canonical trigger is the post-``hermes update`` - stale-module window, where lazy imports read NEW source against OLD cached modules.""" + """Generate and store the model title (daemon-thread target); skips sessions already carrying an + ``llm``/``user`` title (a ``derived`` one is expected — upgrading it is the point). Never lets an + exception escape (the threading excepthook would spray a traceback into the terminal); the canonical + trigger is the post-``hermes update`` window where lazy imports read NEW source against OLD modules.""" try: - # A derived title is expected here — upgrading it is the point. if not session_db or not session_id or _has_upgraded_title(session_db, session_id): return - # This daemon thread starts AFTER the turn's ambient conversation context was reset; - # republish it so the title call carries the same Portal ``conversation=`` tag - # (root-of-lineage) and bills usage to this session. + # This thread starts AFTER the turn's ambient context was reset; republish it so the call carries + # the same Portal ``conversation=`` tag (root-of-lineage) and bills usage to this session. from agent.aux_accounting import set_accounting_context from agent.portal_tags import set_conversation_context - conversation_id = session_id with suppress(Exception): conversation_id = session_db.get_conversation_root(session_id) or session_id @@ -379,9 +363,7 @@ def auto_title_session( user_message, failure_callback=failure_callback, main_runtime=main_runtime, runtime_validator=runtime_validator, ) source = "llm" - if not title: - # The inline attempt declines collisions rather than scan the lineage on the critical - # path; off that path the scan is affordable. + if not title: # the inline attempt declined collisions; off the critical path the lineage scan is affordable title, source = derive_title(user_message), "derived" if not title: return @@ -409,7 +391,7 @@ def _is_real_user_turn(message: Any) -> bool: def _session_is_untitled(session_db, session_id: str) -> bool: - """No title of any provenance. False when it can't tell — no model call per turn for an unreadable title.""" + """No title of any provenance; False when it can't tell (no model call per turn for an unreadable title).""" getter = getattr(session_db, "get_session_title", None) if not callable(getter): return False @@ -430,19 +412,17 @@ def maybe_auto_title( title_callback: Optional[TitleCallback] = None, runtime_validator: Optional[RuntimeValidator] = None, ) -> None: - """Title a session from its opening message: instant inline, then upgraded on a daemon thread. - Call at the START of a turn, before the model is invoked.""" + """Instant inline title, then a daemon-thread upgrade. Call at the START of a turn, before the model.""" if not session_db or not session_id or not user_message: return - # History may be pre- or post-message. Skip only when BOTH past the opening turn AND already named: - # count alone left a machinery-opened session nameless; title alone never titles on an old store. + # History may be pre- or post-message. Skip only when BOTH past the opening turn AND named: count alone + # left a machinery-opened session nameless; title alone never titles on an old store. user_msg_count = sum(1 for m in (conversation_history or []) if _is_real_user_turn(m)) if user_msg_count > 1 and not _session_is_untitled(session_db, session_id): return if not is_titleable_user_message(user_message): return - # Config read after the cheap guards so the file isn't touched every turn. - if not _auto_title_enabled(): + if not _auto_title_enabled(): # config read after the cheap guards so the file isn't touched every turn logger.debug("Auto-title skipped: auxiliary.title_generation.enabled=false") return apply_instant_title(session_db, session_id, user_message, title_callback) diff --git a/agent/trace_upload.py b/agent/trace_upload.py index 2da9d6accd..10a87030d6 100644 --- a/agent/trace_upload.py +++ b/agent/trace_upload.py @@ -1,12 +1,9 @@ -"""Upload a Hermes session transcript to Hugging Face as an agent trace. - -The SQLite session is re-emitted in the **Claude Code JSONL** shape the HF Agent Trace Viewer -auto-detects (https://huggingface.co/docs/hub/agent-traces). Deterministic, zero LLM turns. -Private by default: traces can contain prompts, tool output, local paths and secrets, so the -dataset is created private and every text body goes through the secret redactor (``force=True``) -unless the caller passes ``redact=False``. :func:`upload_session_trace` never raises — it returns a -user-facing status string; programmatic callers use :func:`build_trace_jsonl` + :func:`_do_upload`. -""" +"""Upload a Hermes session transcript to Hugging Face as an agent trace, re-emitted in the **Claude Code +JSONL** shape the HF Agent Trace Viewer auto-detects (https://huggingface.co/docs/hub/agent-traces). +Deterministic, zero LLM turns. Private by default: traces can carry prompts, tool output, local paths and +secrets, so the dataset is created private and every text body passes the secret redactor (``force=True``) +unless ``redact=False``. :func:`upload_session_trace` never raises (returns a user-facing status string); +programmatic callers use :func:`build_trace_jsonl` + :func:`_do_upload`.""" from __future__ import annotations @@ -50,8 +47,7 @@ def _now_iso() -> str: def _redact(text: Any, enabled: bool) -> Any: - """Redact secrets from a string body when enabled (``force=True``: an upload always scrubs - even if log redaction is disabled); non-strings pass through.""" + """Redact a string body when enabled (``force=True``: an upload scrubs even if log redaction is off).""" if not enabled or not isinstance(text, str) or not text: return text try: @@ -162,9 +158,8 @@ _ROLE_RENDERERS: Dict[Any, Tuple[str, Any]] = { def build_trace_jsonl(messages: List[Dict[str, Any]], *, session_id: str, model: str = "", cwd: str = "", redact: bool = True) -> str: - """Render messages as Claude Code JSONL: one line per non-system message (``user``/``tool`` -> - type user, ``assistant`` -> type assistant with text + ``tool_use`` blocks; tool results ride - on user turns as ``tool_result`` keyed by ``tool_call_id``; turns link via ``parentUuid``).""" + """One JSONL line per non-system message: ``user``/``tool`` -> type user (tool results ride on user turns as + ``tool_result`` keyed by ``tool_call_id``), ``assistant`` -> text + ``tool_use`` blocks; turns link via ``parentUuid``.""" lines: List[str] = [] parent: Optional[str] = None base_ts = _now_iso() @@ -201,8 +196,7 @@ def _resolve_hf_token() -> Optional[str]: def _do_upload(jsonl: str, *, token: str, session_id: str, dataset_name: str = DEFAULT_DATASET_NAME, private: bool = True) -> str: - """Create (idempotently) the private dataset and push the trace file. - Returns a user-facing status string. Never raises.""" + """Create the dataset (idempotent) and push the trace file; user-facing status string, never raises.""" with suppress(Exception): # lazy-install unavailable/declined — the import below surfaces the hint from tools import lazy_deps lazy_deps.ensure("tool.trace_upload", prompt=False) @@ -239,8 +233,8 @@ def _do_upload(jsonl: str, *, token: str, session_id: str, dataset_name: str = D def load_session_messages(session_id: str, db_path=None) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: - """Load ``(messages, meta)`` from the SQLite store; ``meta`` is ``{}`` when the session row is - missing (messages may still exist for a live, untitled session).""" + """``(messages, meta)`` from SQLite; ``meta`` is ``{}`` when the session row is missing (a live, untitled + session may still have messages).""" from hermes_state import SessionDB db = SessionDB(db_path=db_path) if db_path else SessionDB() try: @@ -265,8 +259,7 @@ def upload_session_trace( db_path=None, token: Optional[str] = None, ) -> str: - """CLI/gateway entry point: load, convert, upload to the user's private ``{user}/hermes-traces`` - dataset. Returns a status string, never raises.""" + """CLI/gateway entry point: load, convert, upload to ``{user}/hermes-traces``. Status string, never raises.""" if not session_id: return "No active session to upload." token = token or _resolve_hf_token() diff --git a/agent/trajectory.py b/agent/trajectory.py index b7b3fff2d4..50d004d588 100644 --- a/agent/trajectory.py +++ b/agent/trajectory.py @@ -1,5 +1,4 @@ -"""Trajectory saving utilities and static helpers (``_convert_to_trajectory_format`` stays an -AIAgent method — batch_runner.py calls it).""" +"""Trajectory saving + scratchpad helpers (``_convert_to_trajectory_format`` stays an AIAgent method — batch_runner.py calls it).""" import json import logging @@ -22,16 +21,10 @@ def has_incomplete_scratchpad(content: str) -> bool: def save_trajectory(trajectory: List[Dict[str, Any]], model: str, completed: bool, filename: str = None): - """Append a ShareGPT-format trajectory entry to a JSONL file (default trajectory_samples.jsonl / - failed_trajectories.jsonl based on ``completed``).""" + """Append a ShareGPT-format entry to a JSONL file (default trajectory_samples.jsonl / failed_trajectories.jsonl by ``completed``).""" if filename is None: filename = "trajectory_samples.jsonl" if completed else "failed_trajectories.jsonl" - entry = { - "conversations": trajectory, - "timestamp": datetime.now().isoformat(), - "model": model, - "completed": completed, - } + entry = {"conversations": trajectory, "timestamp": datetime.now().isoformat(), "model": model, "completed": completed} try: with open(filename, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") diff --git a/agent/transcript_repair.py b/agent/transcript_repair.py index 479fd67678..3b1c7ae3da 100644 --- a/agent/transcript_repair.py +++ b/agent/transcript_repair.py @@ -1,7 +1,6 @@ -"""Transcript repair for SessionDB batch appends: reconcile in-memory assistant -rows with committed SQLite rows (blank-row in-place update, concurrent-winner -adoption, watermark-compaction clone lookup) and sync markers after commit. -""" +"""Transcript repair for SessionDB batch appends: reconcile in-memory assistant rows with committed SQLite +rows (blank-row in-place update, concurrent-winner adoption, watermark-compaction clone lookup) and sync +markers after commit.""" from __future__ import annotations @@ -18,11 +17,7 @@ def is_content_blank(content: Any) -> bool: if isinstance(content, str): return not content.strip() if isinstance(content, list): - if not content: - return True - return not "".join( - p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text" - ).strip() + return not "".join(p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text").strip() return False @@ -33,54 +28,54 @@ def resolve_and_repair_transcript_batch( encode_content_fn: Callable[[Any], Any], decode_content_fn: Callable[[Any], Any], ) -> List[Dict[str, Any]]: - """Partition a message batch within an active write transaction. - - An assistant message carrying an existing integer ``_row_id`` targets its active SQLite row (or - the active clone a watermark compaction made of it): a blank row is updated in place; a - non-blank one (concurrent winner) has its canonical content adopted without overwrite. - Returns the messages that must be inserted as fresh rows. - """ + """Partition a message batch within an active write transaction. An assistant message carrying an + existing integer ``_row_id`` targets its active SQLite row (or the active clone a watermark compaction + made of it): a blank row is updated in place; a non-blank one (concurrent winner) has its canonical + content adopted without overwrite. Returns the messages that must be inserted as fresh rows.""" inserted_rows: List[Dict[str, Any]] = [] for msg in messages: - repaired = False existing_row_id = msg.get("_row_id") if isinstance(msg, dict) else None + target_row = None if isinstance(existing_row_id, int) and msg.get("role", "unknown") == "assistant": - row = conn.execute( - "SELECT id, role, active, timestamp, content FROM messages " - "WHERE id = ? AND session_id = ?", - (existing_row_id, session_id), - ).fetchone() - target_row = None - if row is not None and row["role"] == "assistant": - if int(row["active"] or 0) == 1: - target_row = row - else: - # Watermark compaction soft-archived the concurrent tail and cloned it. - target_row = conn.execute( - "SELECT id, role, active, timestamp, content FROM messages " - "WHERE session_id = ? AND active = 1 AND role = 'assistant' " - "AND timestamp IS ? AND id != ? " - "ORDER BY id DESC LIMIT 1", - (session_id, row["timestamp"], row["id"]), - ).fetchone() - if target_row is not None: - target_id = int(target_row["id"]) - decoded = decode_content_fn(target_row["content"]) - msg["_row_id"] = target_id - if is_content_blank(decoded): - conn.execute( - "UPDATE messages SET content = ? " - "WHERE id = ? AND session_id = ? AND active = 1", - (encode_content_fn(msg.get("content")), target_id, session_id), - ) - else: - msg["_canonical_content"] = decoded # concurrent winner: adopt, don't overwrite - repaired = True - if not repaired: + target_row = _active_assistant_row(conn, session_id, existing_row_id) + if target_row is None: inserted_rows.append(msg) + continue + target_id = int(target_row["id"]) + decoded = decode_content_fn(target_row["content"]) + msg["_row_id"] = target_id + if is_content_blank(decoded): + conn.execute( + "UPDATE messages SET content = ? " + "WHERE id = ? AND session_id = ? AND active = 1", + (encode_content_fn(msg.get("content")), target_id, session_id), + ) + else: + msg["_canonical_content"] = decoded # concurrent winner: adopt, don't overwrite return inserted_rows +def _active_assistant_row(conn: sqlite3.Connection, session_id: str, row_id: int): + """The active assistant row for ``row_id``, or the active clone a watermark compaction made of it.""" + row = conn.execute( + "SELECT id, role, active, timestamp, content FROM messages " + "WHERE id = ? AND session_id = ?", + (row_id, session_id), + ).fetchone() + if row is None or row["role"] != "assistant": + return None + if int(row["active"] or 0) == 1: + return row + # Watermark compaction soft-archived the concurrent tail and cloned it. + return conn.execute( + "SELECT id, role, active, timestamp, content FROM messages " + "WHERE session_id = ? AND active = 1 AND role = 'assistant' " + "AND timestamp IS ? AND id != ? " + "ORDER BY id DESC LIMIT 1", + (session_id, row["timestamp"], row["id"]), + ).fetchone() + + def sync_flushed_message_markers(batch_msgs: List[Dict[str, Any]], batch_rows: List[Dict[str, Any]]) -> None: """Stamp _DB_PERSISTED_MARKER and sync canonical row ID / content onto live dicts after commit.""" for written, row in zip(batch_msgs, batch_rows):