diff --git a/agent/replay_cleanup.py b/agent/replay_cleanup.py index b23e8bddcd..7490df2197 100644 --- a/agent/replay_cleanup.py +++ b/agent/replay_cleanup.py @@ -18,6 +18,16 @@ from agent.turn_context import drop_stale_api_content logger = logging.getLogger(__name__) +# Orphan-recovery notices: (side-effecting, read-only) for an interrupted block vs a dangling tail. +_INTERRUPTED_NOTICES = ( + "[Orphan recovery: interrupted side-effecting tool may have executed; its effect is UNKNOWN. Inspect state before retrying.]", + "[Orphan recovery: interrupted read-only tool did not complete.]", +) +_DANGLING_NOTICES = ( + "[Orphan recovery: this tool may have executed before Hermes stopped; its effect is UNKNOWN. Inspect current state before retrying.]", + "[Orphan recovery: this read-only tool did not complete and had no effect.]", +) + def is_interrupted_tool_result(content: Any) -> bool: """Return True if a tool result indicates the tool was interrupted.""" @@ -26,9 +36,7 @@ def is_interrupted_tool_result(content: Any) -> bool: lowered = content.lower() if "[command interrupted]" in lowered: return True - if "exit_code" in lowered and ("130" in lowered or "-1" in lowered): - return "interrupt" in lowered - return False + return "exit_code" in lowered and ("130" in lowered or "-1" in lowered) and "interrupt" in lowered def _call_name(call: Dict[str, Any]) -> str: @@ -43,11 +51,11 @@ def _any_side_effecting(calls: List[Dict[str, Any]]) -> bool: return any(tool_may_have_side_effect(_call_name(call)) for call in calls) -def _orphan_recovery(name: str, unknown_text: str, none_text: str) -> tuple: +def _orphan_recovery(name: str, notices: tuple) -> tuple: """(effect_disposition, content) for an interrupted/dangling call named ``name``.""" if tool_may_have_side_effect(name): - return "unknown", unknown_text - return "none", none_text + return "unknown", notices[0] + return "none", notices[1] def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Dict[str, Any]]: @@ -56,7 +64,6 @@ def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Di 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]] = [] i = 0 n = len(agent_history) @@ -64,10 +71,9 @@ def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Di msg = agent_history[i] if msg.get("role") == "assistant" and "tool_calls" in msg: j = i + 1 - tool_results: List[Dict[str, Any]] = [] while j < n and agent_history[j].get("role") == "tool": - tool_results.append(agent_history[j]) j += 1 + tool_results = agent_history[i + 1:j] if tool_results and any(is_interrupted_tool_result(m.get("content", "")) for m in tool_results): calls = msg.get("tool_calls") or [] if _any_side_effecting(calls): @@ -79,28 +85,20 @@ def strip_interrupted_tool_tails(agent_history: List[Dict[str, Any]]) -> List[Di continue recovered = dict(tool_result) name = call_names.get(str(tool_result.get("tool_call_id") or ""), "") - recovered["effect_disposition"], recovered["content"] = _orphan_recovery( - name, - "[Orphan recovery: interrupted side-effecting tool may have " - "executed; its effect is UNKNOWN. Inspect state before retrying.]", - "[Orphan recovery: interrupted read-only tool did not complete.]", - ) + recovered["effect_disposition"], recovered["content"] = _orphan_recovery(name, _INTERRUPTED_NOTICES) cleaned.append(recovered) else: logger.debug( - "Stripping interrupted read-only assistant→tool replay block " - "(indices %d–%d, tool_results=%d)", + "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", "")): logger.debug("Stripping orphan interrupted tool result from replay history") - i += 1 - continue - cleaned.append(msg) + else: + cleaned.append(msg) i += 1 - return cleaned @@ -111,26 +109,18 @@ def strip_dangling_tool_call_tail(agent_history: List[Dict[str, Any]]) -> List[D ones get synthetic UNKNOWN-effect results.""" if not agent_history: return agent_history - last = agent_history[-1] if not (isinstance(last, dict) and last.get("role") == "assistant" and last.get("tool_calls")): return agent_history - tool_calls = last.get("tool_calls") or [] if _any_side_effecting(tool_calls): recovered = list(agent_history) for call in tool_calls: - name = str((call.get("function") or {}).get("name") or "unknown") - disposition, content = _orphan_recovery( - name, - "[Orphan recovery: this tool may have executed before Hermes stopped; " - "its effect is UNKNOWN. Inspect current state before retrying.]", - "[Orphan recovery: this read-only tool did not complete and had no effect.]", - ) + name = _call_name(call) or "unknown" + disposition, content = _orphan_recovery(name, _DANGLING_NOTICES) recovered.append(make_tool_result_message(name, content, _call_id(call), effect_disposition=disposition)) logger.warning("Recovered dangling side-effecting tool call(s) as UNKNOWN instead of erasing them") return recovered - logger.debug("Stripping dangling unanswered read-only assistant(tool_calls) tail (%d call(s))", len(tool_calls)) return agent_history[:-1] @@ -149,20 +139,11 @@ def sanitize_replay_history(agent_history: List[Dict[str, Any]]) -> List[Dict[st _DANGEROUS_CONFIRMATION_EXPIRY_SECONDS = 60.0 # Confirmation phrases that unlock destructive host actions; case-insensitive substring match so -# trailing punctuation / extra context still matches. +# 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", - # i18n variants observed in the original incident - "確認強制重開機", - "確認強制重開", - "確認重啟", + "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 @@ -193,26 +174,20 @@ def strip_stale_dangerous_confirmations( without a timestamp (legacy transcripts, test scaffolding) 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 ( - ts is not None - and is_dangerous_confirmation(msg.get("content", "")) - 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) + if ts is None or not is_dangerous_confirmation(msg.get("content", "")) or (now - float(ts)) <= expiry_seconds: + cleaned.append(msg) continue - cleaned.append(msg) + 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) return cleaned diff --git a/agent/trace_upload.py b/agent/trace_upload.py index 6bda045491..2da9d6accd 100644 --- a/agent/trace_upload.py +++ b/agent/trace_upload.py @@ -36,6 +36,7 @@ _NO_TOKEN_MESSAGE = ( " HF_TOKEN=hf_xxxxxxxxxxxxxxxxxxxx\n" "3. Run /upload-trace again (or `hermes trace upload`)." ) +_TOKEN_ENV_VARS = ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN") class TraceRedactionError(RuntimeError): @@ -80,11 +81,9 @@ def _content_to_blocks(content: Any, redact: bool) -> List[Dict[str, Any]]: """Normalize a message ``content`` field into Anthropic content blocks.""" if content is None: return [] - if isinstance(content, str): - return [_text_block(content, redact)] if isinstance(content, list): return [_part_to_block(part, redact) for part in content] - return [_text_block(json.dumps(content), redact)] + return [_text_block(content if isinstance(content, str) else json.dumps(content), redact)] def _parse_tool_args(raw_args: Any) -> Dict[str, Any]: @@ -152,10 +151,7 @@ def _tool_result_message(msg: Dict[str, Any], model: str, redact: bool) -> Dict[ def _user_message(msg: Dict[str, Any], model: str, redact: bool) -> Dict[str, Any]: content = msg.get("content") - return { - "role": "user", - "content": _redact(content, redact) if isinstance(content, str) else _content_to_blocks(content, redact), - } + return {"role": "user", "content": _redact(content, redact) if isinstance(content, str) else _content_to_blocks(content, redact)} # role -> (Claude Code line type, message builder). Unknown roles render as user. @@ -165,14 +161,7 @@ _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: +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``).""" @@ -180,7 +169,6 @@ def build_trace_jsonl( parent: Optional[str] = None base_ts = _now_iso() git_branch = _git_branch(cwd) - for msg in messages: role = msg.get("role") if role == "system": @@ -202,7 +190,6 @@ def build_trace_jsonl( } lines.append(json.dumps(entry, ensure_ascii=False)) parent = turn_uuid - return "\n".join(lines) + ("\n" if lines else "") @@ -210,21 +197,10 @@ def build_trace_jsonl( def _resolve_hf_token() -> Optional[str]: """Return the user's Hugging Face token from the usual env vars.""" - for var in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN"): - val = (os.getenv(var) or "").strip() - if val: - return val - return None + return next((val for var in _TOKEN_ENV_VARS if (val := (os.getenv(var) or "").strip())), None) -def _do_upload( - jsonl: str, - *, - token: str, - session_id: str, - dataset_name: str = DEFAULT_DATASET_NAME, - private: bool = True, -) -> 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.""" with suppress(Exception): # lazy-install unavailable/declined — the import below surfaces the hint @@ -234,38 +210,30 @@ def _do_upload( from huggingface_hub import HfApi except ImportError: return "Hugging Face upload needs the `huggingface_hub` package (`pip install huggingface_hub`)." - api = HfApi(token=token) try: who = api.whoami() user = who.get("name") if isinstance(who, dict) else None except Exception as e: logger.warning("HF whoami failed: %s", e) - return ("Your Hugging Face token was rejected (whoami failed). " - "Make sure it has WRITE access and isn't expired.") + return "Your Hugging Face token was rejected (whoami failed). Make sure it has WRITE access and isn't expired." if not user: return "Could not resolve your Hugging Face username from the token." - repo_id = f"{user}/{dataset_name}" try: api.create_repo(repo_id=repo_id, repo_type="dataset", private=private, exist_ok=True) except Exception as e: logger.warning("HF create_repo failed for %s: %s", repo_id, e) return f"Could not create/access dataset {repo_id}: {e}" - path_in_repo = f"sessions/{session_id}.jsonl" try: api.upload_file( - path_or_fileobj=jsonl.encode("utf-8"), - path_in_repo=path_in_repo, - repo_id=repo_id, - repo_type="dataset", + path_or_fileobj=jsonl.encode("utf-8"), path_in_repo=path_in_repo, repo_id=repo_id, repo_type="dataset", commit_message=f"add session trace {session_id}", ) except Exception as e: logger.warning("HF upload_file failed for %s: %s", repo_id, e) return f"Upload to Hugging Face failed: {e}" - return (f"Uploaded -> https://huggingface.co/datasets/{repo_id}/blob/main/{path_in_repo}\n" f"View in the trace viewer: https://huggingface.co/datasets/{repo_id}") @@ -301,11 +269,9 @@ def upload_session_trace( dataset. Returns a status string, never raises.""" if not session_id: return "No active session to upload." - token = token or _resolve_hf_token() if not token: return _NO_TOKEN_MESSAGE - try: messages, meta = load_session_messages(session_id, db_path=db_path) except Exception as e: @@ -319,5 +285,4 @@ def upload_session_trace( return _REDACTION_BLOCKED_MESSAGE if not jsonl.strip(): return "No transcript content to upload for this session." - return _do_upload(jsonl, token=token, session_id=session_id, dataset_name=dataset_name, private=private)