ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(agent/trace_upload,replay_cleanup): notice tables, token env tuple, single-branch tail loops

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:50:13 -07:00
parent cd1cec9945
commit 04a2c422f6
2 changed files with 45 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/replay_cleanup.py
+37 -62
View File
@@ -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
agent/trace_upload.py
+8 -43
View File
@@ -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)