diff --git a/agent/replay_cleanup.py b/agent/replay_cleanup.py index f1a18a19c9..3c7c408499 100644 --- a/agent/replay_cleanup.py +++ b/agent/replay_cleanup.py @@ -1,11 +1,10 @@ """Replay-history sanitization shared across resume code paths. -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 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). """ from __future__ import annotations @@ -54,12 +53,11 @@ def _orphan_recovery(name: str, unknown_text: str, none_text: str) -> tuple: def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Strip interrupted assistant→tool sequences from replay history. - The interrupted block is not necessarily the final tail (a queued real user - message may follow it), so every contiguous assistant(tool_calls)+tool-result - block containing an interrupted result is handled; successful sequences stay - intact. 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 already have happened and erasing it would hide that. + The interrupted block is not necessarily the final tail (a queued real user message may + follow it), so every contiguous assistant(tool_calls)+tool-result block containing an + interrupted result is handled; successful sequences stay intact. 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 already have happened. """ if not agent_history: return agent_history @@ -93,13 +91,12 @@ def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Di "[Orphan recovery: interrupted read-only tool did not complete.]", ) cleaned.append(recovered) - i = j - continue - logger.debug( - "Stripping interrupted read-only assistant→tool replay block " - "(indices %d–%d, tool_results=%d)", - i, j - 1, len(tool_results), - ) + else: + logger.debug( + "Stripping interrupted read-only assistant→tool replay block " + "(indices %d–%d, tool_results=%d)", + i, j - 1, len(tool_results), + ) i = j continue if msg.get("role") == "tool" and is_interrupted_tool_result(msg.get("content", "")): @@ -115,13 +112,11 @@ 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)`` block left with NO answers. - A tool call that kills the gateway process itself (``docker restart``, - ``hermes gateway restart``) is SIGKILLed mid-call, before any tool result or - the orderly shutdown rewind; the persisted tail is the assistant message with - zero matching ``tool`` rows, which ``strip_interrupted_tool_tails`` cannot - detect (no result to inspect). Only acts when the tail has NO tool answers — - a partially answered block still resumes. Read-only tails are dropped; - side-effecting ones get synthetic UNKNOWN-effect results instead of erasure. + A tool call that kills the gateway process itself (``docker restart``, ``hermes gateway + restart``) is SIGKILLed mid-call, before any tool result or the orderly shutdown rewind; the + persisted tail has zero matching ``tool`` rows, which ``strip_interrupted_tool_tails`` cannot + detect. Only acts when the tail has NO tool answers — a partially answered block still + resumes. Read-only tails are dropped; side-effecting ones get synthetic UNKNOWN-effect results. """ if not agent_history: return agent_history @@ -155,23 +150,20 @@ 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 replay-tail strippers in canonical order (interrupted blocks, then dangling tail). + Returns the same list object when nothing is stripped.""" if not agent_history: return agent_history return strip_dangling_tool_call_tail(strip_interrupted_tool_tails(agent_history)) -# ────────────────────────────────────────────────────────────────────── -# Stale dangerous-confirmation text expiry -# ────────────────────────────────────────────────────────────────────── +# --- Stale dangerous-confirmation text expiry --- -# Short on purpose: a dangerous confirmation must not survive any restart or -# resume gap. The user can always re-confirm. +# 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. +# Confirmation phrases that unlock destructive host actions; case-insensitive substring match so +# trailing punctuation / extra context still matches. _DANGEROUS_CONFIRMATION_PATTERNS: tuple = ( "confirm forced restart", "confirm forced reboot", @@ -187,8 +179,8 @@ _DANGEROUS_CONFIRMATION_PATTERNS: tuple = ( "確認重啟", ) -# Redacting in place (rather than deleting the message) preserves strict -# user/assistant role alternation in the replayed history. +# Redacting in place (rather than deleting the message) preserves strict user/assistant role +# alternation in the replayed history. _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 " @@ -212,39 +204,34 @@ def strip_stale_dangerous_confirmations( ) -> List[Dict[str, Any]]: """Expire stale dangerous-confirmation text in user messages. - If a host restart killed the gateway before the tool result was written, the - user's confirmation phrase survives in the transcript; a casual "are you - there?" minutes later can read to the model as a fresh re-confirmation and - re-execute the destructive action. Expired confirmations are REDACTED IN - PLACE (deleting the message would leave two consecutive assistant turns). - Messages without a timestamp (legacy transcripts, test scaffolding) and - confirmations still inside the expiry window are left untouched. + If a host restart killed the gateway before the tool result was written, the user's + confirmation phrase survives in the transcript; a casual "are you there?" minutes later can + read to the model as a fresh re-confirmation. Expired confirmations are REDACTED IN PLACE. + Messages without a timestamp (legacy transcripts, test scaffolding) and confirmations still + inside the expiry window are left untouched. """ if not agent_history: return agent_history cleaned: List[Dict[str, Any]] = [] for msg in agent_history: + ts = msg.get("timestamp") if isinstance(msg, dict) and msg.get("role") == "user" else None if ( - isinstance(msg, dict) - and msg.get("role") == "user" + ts is not None and is_dangerous_confirmation(msg.get("content", "")) + and (now - float(ts)) > expiry_seconds ): - ts = msg.get("timestamp") - if ts is not None and (now - float(ts)) > expiry_seconds: - logger.debug( - "Redacting stale dangerous-confirmation text in user " - "message (age=%.1fs, expiry=%.1fs): %r", - now - float(ts), - expiry_seconds, - (msg.get("content") or "")[:80], - ) - 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. - drop_stale_api_content(redacted) - cleaned.append(redacted) - continue + logger.debug( + "Redacting stale dangerous-confirmation text in user " + "message (age=%.1fs, expiry=%.1fs): %r", + now - float(ts), expiry_seconds, (msg.get("content") or "")[:80], + ) + 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. + drop_stale_api_content(redacted) + cleaned.append(redacted) + continue cleaned.append(msg) return cleaned diff --git a/agent/session_activity.py b/agent/session_activity.py index 2a644dd1d5..b7a161bc65 100644 --- a/agent/session_activity.py +++ b/agent/session_activity.py @@ -1,9 +1,8 @@ """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=``. +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=``. """ from __future__ import annotations @@ -14,11 +13,10 @@ 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 the heartbeat is an observation-only -# projection that never justifies extra write pressure. Deliberately a code -# constant (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 turn it into 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 @@ -36,9 +34,7 @@ class ActivityProvenance(str, Enum): def bound_activity_description(description: Optional[str]) -> str: """Clamp free-form activity text to the shared description budget.""" text = (description or "").strip() - if len(text) <= ACTIVITY_DESCRIPTION_MAX: - return text - return text[: ACTIVITY_DESCRIPTION_MAX - 1] + "…" + return text if len(text) <= ACTIVITY_DESCRIPTION_MAX else text[: ACTIVITY_DESCRIPTION_MAX - 1] + "…" def normalize_activity_provenance(provenance: Optional[ActivityProvenance | str]) -> ActivityProvenance: @@ -52,8 +48,8 @@ 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 durable persist rate-limit so the next stamp writes through (terminal compression + labels must not stay stuck on mid-compress text).""" try: agent._session_activity_last_persist_mono = 0.0 except Exception: diff --git a/agent/trace_upload.py b/agent/trace_upload.py index 276e8f6341..987d28ef93 100644 --- a/agent/trace_upload.py +++ b/agent/trace_upload.py @@ -1,16 +1,11 @@ """Upload a Hermes session transcript to Hugging Face as an agent trace. -The SQLite session (``hermes_state.SessionDB``) is re-emitted in the **Claude -Code JSONL** shape, one of the formats the HF Agent Trace Viewer auto-detects -(docs: https://huggingface.co/docs/hub/agent-traces). - -* **Zero LLM turn.** Deterministic export; ``hermes trace upload`` calls - :func:`upload_session_trace` directly. -* **Private by default.** Traces can contain prompts, tool output, local paths and - secrets: the dataset is created private and every text body goes through the - secret redactor (``force=True``) unless the caller passes ``redact=False``. -* **Never raises.** Returns a user-facing status string. Programmatic callers - wanting the URL use :func:`build_trace_jsonl` + :func:`_do_upload` directly. +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`. """ from __future__ import annotations @@ -46,17 +41,15 @@ class TraceRedactionError(RuntimeError): """Raised when a trace cannot be safely redacted before upload.""" -# --------------------------------------------------------------------------- -# Conversion: Hermes OpenAI-format messages -> Claude Code JSONL -# --------------------------------------------------------------------------- +# --- Conversion: Hermes OpenAI-format messages -> Claude Code JSONL --- def _now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" def _redact(text: Any, enabled: bool) -> Any: - """Redact secrets from a string body when enabled; non-strings pass through. - ``force=True``: an upload always scrubs even if log redaction is disabled.""" + """Redact secrets from a string body when enabled (``force=True``: an upload always scrubs + even if log redaction is disabled); non-strings pass through.""" if not enabled or not isinstance(text, str) or not text: return text try: @@ -183,10 +176,10 @@ def build_trace_jsonl( ) -> str: """Render Hermes conversation messages as Claude Code JSONL text. - Each non-system message becomes one line: ``user``/``tool`` -> ``{"type": - "user"}``, ``assistant`` -> ``{"type": "assistant"}`` with text + ``tool_use`` - blocks. Tool results ride on user turns as a ``tool_result`` block keyed by - ``tool_call_id``; turns link via ``uuid`` / ``parentUuid``. + Each non-system message becomes one line: ``user``/``tool`` -> ``{"type": "user"}``, + ``assistant`` -> ``{"type": "assistant"}`` with text + ``tool_use`` blocks. Tool results ride + on user turns as a ``tool_result`` block keyed by ``tool_call_id``; turns link via ``uuid`` / + ``parentUuid``. """ lines: List[str] = [] parent: Optional[str] = None @@ -218,9 +211,7 @@ def build_trace_jsonl( return "\n".join(lines) + ("\n" if lines else "") -# --------------------------------------------------------------------------- -# Upload -# --------------------------------------------------------------------------- +# --- Upload --- def _resolve_hf_token() -> Optional[str]: """Return the user's Hugging Face token from the usual env vars.""" @@ -245,8 +236,7 @@ def _do_upload( from tools import lazy_deps lazy_deps.ensure("tool.trace_upload", prompt=False) except Exception: - # Lazy-install unavailable/declined — the import below surfaces the hint. - pass + pass # lazy-install unavailable/declined — the import below surfaces the hint try: from huggingface_hub import HfApi except ImportError: @@ -288,8 +278,8 @@ def _do_upload( 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).""" + """Load ``(messages, meta)`` from the SQLite store; ``meta`` is ``{}`` when the session row is + missing (messages may still exist for a live, untitled session).""" from hermes_state import SessionDB db = SessionDB(db_path=db_path) if db_path else SessionDB() try: @@ -314,8 +304,8 @@ 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 the user's private ``{user}/hermes-traces`` + dataset. Returns a status string, never raises.""" if not session_id: return "No active session to upload." diff --git a/agent/trajectory.py b/agent/trajectory.py index fc43ce74f9..b7b3fff2d4 100644 --- a/agent/trajectory.py +++ b/agent/trajectory.py @@ -1,8 +1,5 @@ -"""Trajectory saving utilities and static helpers. - -_convert_to_trajectory_format stays as an AIAgent method (batch_runner.py calls -agent._convert_to_trajectory_format); only static helpers and file-write logic live here. -""" +"""Trajectory saving utilities and static helpers (``_convert_to_trajectory_format`` stays an +AIAgent method — batch_runner.py calls it).""" import json import logging @@ -25,8 +22,8 @@ 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 trajectory entry to a JSONL file (default trajectory_samples.jsonl / + failed_trajectories.jsonl based on ``completed``).""" if filename is None: filename = "trajectory_samples.jsonl" if completed else "failed_trajectories.jsonl" entry = { diff --git a/agent/transcript_repair.py b/agent/transcript_repair.py index 7da566ead9..479fd67678 100644 --- a/agent/transcript_repair.py +++ b/agent/transcript_repair.py @@ -35,18 +35,16 @@ def resolve_and_repair_transcript_batch( ) -> List[Dict[str, Any]]: """Partition a message batch within an active write transaction. - For assistant messages carrying an existing integer `_row_id`: - - Checks for an active target row or watermark compaction clone in SQLite. - - If blank, updates the row in-place with new content. - - If already non-blank (concurrent winner), adopts canonical content without overwrite. - - Returns the list of messages that must be inserted as fresh rows. + 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: - role = msg.get("role", "unknown") if isinstance(msg, dict) else "unknown" - existing_row_id = msg.get("_row_id") if isinstance(msg, dict) else None repaired = False - if role == "assistant" and isinstance(existing_row_id, int): + existing_row_id = msg.get("_row_id") if isinstance(msg, dict) else 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 = ?", @@ -57,35 +55,26 @@ def resolve_and_repair_transcript_batch( if int(row["active"] or 0) == 1: target_row = row else: - # Watermark compaction soft-archived the concurrent tail - # and cloned it. Find the active clone. - clone = conn.execute( + # 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 clone is not None: - target_row = clone if target_row is not None: target_id = int(target_row["id"]) - raw_content = target_row["content"] - decoded = decode_content_fn(raw_content) + decoded = decode_content_fn(target_row["content"]) + msg["_row_id"] = target_id if is_content_blank(decoded): - encoded = encode_content_fn(msg.get("content")) conn.execute( "UPDATE messages SET content = ? " "WHERE id = ? AND session_id = ? AND active = 1", - (encoded, target_id, session_id), + (encode_content_fn(msg.get("content")), target_id, session_id), ) - if isinstance(msg, dict): - msg["_row_id"] = target_id else: - # Concurrent winner: adopt canonical content without overwrite - if isinstance(msg, dict): - msg["_row_id"] = target_id - msg["_canonical_content"] = decoded + msg["_canonical_content"] = decoded # concurrent winner: adopt, don't overwrite repaired = True if not repaired: inserted_rows.append(msg)