mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-05-21 05:11:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

chore: ruff auto-fix PLR6201 — tuple → set in membership tests (#23937)

Browse source 
Replace  with  for all literal-tuple
membership tests. Set lookup is O(1) vs O(n) for tuple — consistent
micro-optimization across the codebase.

608 instances fixed via `ruff --fix --unsafe-fixes`, 0 remaining.
133 files, +626/-626 (net zero).
This commit is contained in:
kshitij 2026-05-11 11:13:25 -07:00 • committed by GitHub
parent 8c11710314
commit 2ec8d2b42f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
133 changed files with 626 additions and 626 deletions
Download patch file Download diff file Expand all files Collapse all files

14
tools/approval.py
View file

@ -759,13 +759,13 @@ def prompt_dangerous_approval(command: str, description: str,
return "deny"
choice = result["choice"]
if choice in ('o', 'once'):
if choice in {'o', 'once'}:
print(t("approval.allowed_once"))
return "once"
elif choice in ('s', 'session'):
elif choice in {'s', 'session'}:
print(t("approval.allowed_session"))
return "session"
elif choice in ('a', 'always'):
elif choice in {'a', 'always'}:
if not allow_permanent:
print(t("approval.allowed_session"))
return "session"
@ -831,7 +831,7 @@ def _get_cron_approval_mode() -> str:
from hermes_cli.config import load_config
config = load_config()
mode = str(cfg_get(config, "approvals", "cron_mode", default="deny")).lower().strip()
if mode in ("approve", "off", "allow", "yes"):
if mode in {"approve", "off", "allow", "yes"}:
return "approve"
return "deny"
except Exception:
@ -900,7 +900,7 @@ def check_dangerous_command(command: str, env_type: str,
Returns:
{"approved": True/False, "message": str or None, ...}
"""
if env_type in ("docker", "singularity", "modal", "daytona", "vercel_sandbox"):
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
return {"approved": True, "message": None}
# Hardline floor: commands with no recovery path (rm -rf /, mkfs, dd
@ -1025,7 +1025,7 @@ def check_all_command_guards(command: str, env_type: str,
other was shown to the user.
"""
# Skip containers for both checks
if env_type in ("docker", "singularity", "modal", "daytona", "vercel_sandbox"):
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
return {"approved": True, "message": None}
# Hardline floor: unconditional block for catastrophic commands
@ -1104,7 +1104,7 @@ def check_all_command_guards(command: str, env_type: str,
# Previously, tirith "block" was a hard block with no approval prompt.
# Now both block and warn go through the approval flow so users can
# inspect the explanation and approve if they understand the risk.
if tirith_result["action"] in ("block", "warn"):
if tirith_result["action"] in {"block", "warn"}:
findings = tirith_result.get("findings") or []
rule_id = findings[0].get("rule_id", "unknown") if findings else "unknown"
tirith_key = f"tirith:{rule_id}"

2
tools/browser_providers/browser_use.py
View file

@ -184,7 +184,7 @@ class BrowserUseProvider(CloudBrowserProvider):
json={"action": "stop"},
timeout=10,
)
if response.status_code in (200, 201, 204):
if response.status_code in {200, 201, 204}:
logger.debug("Successfully closed Browser Use session %s", session_id)
return True
else:

2
tools/browser_providers/browserbase.py
View file

@ -180,7 +180,7 @@ class BrowserbaseProvider(CloudBrowserProvider):
},
timeout=10,
)
if response.status_code in (200, 201, 204):
if response.status_code in {200, 201, 204}:
logger.debug("Successfully closed Browserbase session %s", session_id)
return True
else:

2
tools/browser_providers/firecrawl.py
View file

@ -79,7 +79,7 @@ class FirecrawlProvider(CloudBrowserProvider):
headers=self._headers(),
timeout=10,
)
if response.status_code in (200, 201, 204):
if response.status_code in {200, 201, 204}:
logger.debug("Successfully closed Firecrawl session %s", session_id)
return True
else:

6
tools/browser_supervisor.py
View file

@ -412,7 +412,7 @@ class CDPSupervisor:
``{"ok": False, "error": "..."}`` on a recoverable error (no dialog,
ambiguous dialog_id, supervisor inactive).
"""
if action not in ("accept", "dismiss"):
if action not in {"accept", "dismiss"}:
return {"ok": False, "error": f"action must be 'accept' or 'dismiss', got {action!r}"}
with self._state_lock:
@ -1206,7 +1206,7 @@ class CDPSupervisor:
info = params.get("targetInfo") or {}
sid = params.get("sessionId")
target_type = info.get("type")
if not sid or target_type not in ("iframe", "worker"):
if not sid or target_type not in {"iframe", "worker"}:
return
self._child_sessions[sid] = {"info": info, "type": target_type}
@ -1290,7 +1290,7 @@ class CDPSupervisor:
event = ConsoleEvent(ts=time.time(), level="exception", text=text, url=url)
else:
raw_level = str(params.get("type") or "log")
level = "error" if raw_level in ("error", "assert") else (
level = "error" if raw_level in {"error", "assert"} else (
"warning" if raw_level == "warning" else "log"
)
args = params.get("args") or []

4
tools/browser_tool.py
View file

@ -918,7 +918,7 @@ def _url_is_private(url: str) -> bool:
# Hostname — must resolve to confirm it's private (bare "localhost"
# resolves to 127.0.0.1 via /etc/hosts). Short-circuit on obvious
# names to avoid a DNS hop.
if hostname in ("localhost",) or hostname.endswith(".localhost"):
if hostname in {"localhost",} or hostname.endswith(".localhost"):
return True
if hostname.endswith(".local") or hostname.endswith(".lan") or hostname.endswith(".internal"):
return True
@ -2499,7 +2499,7 @@ def browser_scroll(direction: str, task_id: Optional[str] = None) -> str:
JSON string with scroll result
"""
# Validate direction
if direction not in ["up", "down"]:
if direction not in {"up", "down"}:
return json.dumps({
"success": False,
"error": f"Invalid direction '{direction}'. Use 'up' or 'down'."

2
tools/checkpoint_manager.py
View file

@ -639,7 +639,7 @@ class CheckpointManager:
abs_dir = str(_normalize_path(working_dir))
# Skip root, home, and other overly broad directories
if abs_dir in ("/", str(Path.home())):
if abs_dir in {"/", str(Path.home())}:
logger.debug("Checkpoint skipped: directory too broad (%s)", abs_dir)
return False

2
tools/code_execution_tool.py
View file

@ -612,7 +612,7 @@ def _get_or_create_env(task_id: str):
cwd = overrides.get("cwd") or config["cwd"]
container_config = None
if env_type in ("docker", "singularity", "modal", "daytona", "vercel_sandbox"):
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),

2
tools/computer_use/cua_backend.py
View file

@ -673,5 +673,5 @@ def _parse_element(d: Dict[str, Any]) -> UIElement:
pid=int(d.get("pid", 0) or 0),
window_id=int(d.get("windowId", 0) or 0),
attributes={k: v for k, v in d.items()
if k not in ("index", "role", "label", "bounds", "app", "pid", "windowId")},
if k not in {"index", "role", "label", "bounds", "app", "pid", "windowId"}},
)

8
tools/computer_use/tool.py
View file

@ -131,7 +131,7 @@ def _get_backend() -> ComputerUseBackend:
with _backend_lock:
if _backend is None:
backend_name = os.environ.get("HERMES_COMPUTER_USE_BACKEND", "cua").lower()
if backend_name in ("cua", "cua-driver", ""):
if backend_name in {"cua", "cua-driver", ""}:
from tools.computer_use.cua_backend import CuaDriverBackend
_backend = CuaDriverBackend()
elif backend_name == "noop": # pragma: no cover
@ -286,7 +286,7 @@ def _request_approval(action: str, args: Dict[str, Any]) -> Optional[str]:
def _summarize_action(action: str, args: Dict[str, Any]) -> str:
if action in ("click", "double_click", "right_click", "middle_click"):
if action in {"click", "double_click", "right_click", "middle_click"}:
if args.get("element") is not None:
return f"{action} element #{args['element']}"
coord = args.get("coordinate")
@ -314,7 +314,7 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
if action == "capture":
mode = str(args.get("mode", "som"))
if mode not in ("som", "vision", "ax"):
if mode not in {"som", "vision", "ax"}:
return json.dumps({"error": f"bad mode {mode!r}; use som|vision|ax"})
cap = backend.capture(mode=mode, app=args.get("app"))
return _capture_response(cap)
@ -335,7 +335,7 @@ def _dispatch(backend: ComputerUseBackend, action: str, args: Dict[str, Any]) ->
res = backend.focus_app(app, raise_window=bool(args.get("raise_window")))
return _maybe_follow_capture(backend, res, capture_after)
if action in ("click", "double_click", "right_click", "middle_click"):
if action in {"click", "double_click", "right_click", "middle_click"}:
button = args.get("button")
click_count = 1
if action == "double_click":

8
tools/delegate_tool.py
View file

@ -315,7 +315,7 @@ def _normalize_role(r: Optional[str]) -> str:
if r is None or not r:
return "leaf"
r_norm = str(r).strip().lower()
if r_norm in ("leaf", "orchestrator"):
if r_norm in {"leaf", "orchestrator"}:
return r_norm
logger.warning("Unknown delegate_task role=%r, coercing to 'leaf'", r)
return "leaf"
@ -437,7 +437,7 @@ def _get_orchestrator_enabled() -> bool:
return val
# Accept "true"/"false" strings from YAML that doesn't auto-coerce.
if isinstance(val, str):
return val.strip().lower() in ("true", "1", "yes", "on")
return val.strip().lower() in {"true", "1", "yes", "on"}
return True
@ -2271,9 +2271,9 @@ def delegate_task(
# total as "none" when the parent itself hadn't billed any calls
# yet (rare but possible when the parent's only action this turn
# was delegate_task).
if getattr(parent_agent, "session_cost_source", "none") in (None, "", "none"):
if getattr(parent_agent, "session_cost_source", "none") in {None, "", "none"}:
parent_agent.session_cost_source = "subagent"
if getattr(parent_agent, "session_cost_status", "unknown") in (None, "", "unknown"):
if getattr(parent_agent, "session_cost_status", "unknown") in {None, "", "unknown"}:
parent_agent.session_cost_status = "estimated"
except Exception:
logger.debug("Subagent cost rollup failed", exc_info=True)

4
tools/environments/daytona.py
View file

@ -124,7 +124,7 @@ class DaytonaEnvironment(BaseEnvironment):
home = self._sandbox.process.exec("echo $HOME").result.strip()
if home:
self._remote_home = home
if requested_cwd in ("~", "/home/daytona"):
if requested_cwd in {"~", "/home/daytona"}:
self.cwd = home
except Exception:
pass
@ -195,7 +195,7 @@ class DaytonaEnvironment(BaseEnvironment):
def _ensure_sandbox_ready(self) -> None:
"""Restart sandbox if it was stopped (e.g., by a previous interrupt)."""
self._sandbox.refresh_data()
if self._sandbox.state in (self._SandboxState.STOPPED, self._SandboxState.ARCHIVED):
if self._sandbox.state in {self._SandboxState.STOPPED, self._SandboxState.ARCHIVED}:
self._sandbox.start()
logger.info("Daytona: restarted sandbox %s", self._sandbox.id)

4
tools/environments/vercel_sandbox.py
View file

@ -254,7 +254,7 @@ class VercelSandboxEnvironment(BaseEnvironment):
self.init_session()
def _build_create_params(self, *, cpu: float, memory: int, disk: int) -> _SandboxCreateParams:
if disk not in (0, _DEFAULT_CONTAINER_DISK_MB):
if disk not in {0, _DEFAULT_CONTAINER_DISK_MB}:
raise ValueError(
"Vercel Sandbox does not support configurable container_disk. "
"Use the default shared setting."
@ -336,7 +336,7 @@ class VercelSandboxEnvironment(BaseEnvironment):
if requested_cwd == "~":
self.cwd = self._remote_home
elif requested_cwd in ("", DEFAULT_VERCEL_CWD):
elif requested_cwd in {"", DEFAULT_VERCEL_CWD}:
self.cwd = self._workspace_root
else:
self.cwd = requested_cwd

4
tools/file_operations.py
View file

@ -1244,7 +1244,7 @@ class ShellFileOperations(FileOperations):
search_root = Path(path)
has_hidden_path_ancestor = any(
part not in (".", "..") and part.startswith(".")
part not in {".", ".."} and part.startswith(".")
for part in search_root.parts
)
@ -1305,7 +1305,7 @@ class ShellFileOperations(FileOperations):
rel_parts = Path(file_path).resolve().relative_to(normalized_root).parts
except ValueError:
rel_parts = Path(file_path).parts
if any(part not in (".", "..") and part.startswith(".") for part in rel_parts):
if any(part not in {".", ".."} and part.startswith(".") for part in rel_parts):
continue
filtered_files.append(file_path)
files = filtered_files[offset:offset + limit]

2
tools/file_tools.py
View file

@ -380,7 +380,7 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
logger.info("Creating new %s environment for task %s...", env_type, task_id[:8])
container_config = None
if env_type in ("docker", "singularity", "modal", "daytona", "vercel_sandbox"):
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),

2
tools/image_generation_tool.py
View file

@ -575,7 +575,7 @@ def _build_fal_payload(
payload: Dict[str, Any] = dict(meta.get("defaults", {}))
payload["prompt"] = (prompt or "").strip()
if size_style in ("image_size_preset", "gpt_literal"):
if size_style in {"image_size_preset", "gpt_literal"}:
payload["image_size"] = sizes[aspect]
elif size_style == "aspect_ratio":
payload["aspect_ratio"] = sizes[aspect]

6
tools/kanban_tools.py
View file

@ -160,7 +160,7 @@ def _normalize_profile(value: Any) -> Optional[str]:
if value is None:
return None
text = str(value).strip()
if not text or text.lower() in ("none", "-", "null"):
if not text or text.lower() in {"none", "-", "null"}:
return None
return text
@ -172,9 +172,9 @@ def _parse_bool_arg(args: dict, name: str, *, default: bool = False):
if isinstance(value, bool):
return value, None
text = str(value).strip().lower()
if text in ("true", "1", "yes"):
if text in {"true", "1", "yes"}:
return True, None
if text in ("false", "0", "no"):
if text in {"false", "0", "no"}:
return False, None
return default, f"{name} must be a boolean or 'true'/'false'"

2
tools/memory_tool.py
View file

@ -477,7 +477,7 @@ def memory_tool(
if store is None:
return tool_error("Memory is not available. It may be disabled in config or this environment.", success=False)
if target not in ("memory", "user"):
if target not in {"memory", "user"}:
return tool_error(f"Invalid target '{target}'. Use 'memory' or 'user'.", success=False)
if action == "add":

4
tools/osv_check.py
View file

@ -65,9 +65,9 @@ def check_package_for_malware(
def _infer_ecosystem(command: str) -> Optional[str]:
"""Infer package ecosystem from the command name."""
base = os.path.basename(command).lower()
if base in ("npx", "npx.cmd"):
if base in {"npx", "npx.cmd"}:
return "npm"
if base in ("uvx", "uvx.cmd", "pipx"):
if base in {"uvx", "uvx.cmd", "pipx"}:
return "PyPI"
return None

4
tools/patch_parser.py
View file

@ -263,7 +263,7 @@ def _validate_operations(
simulated = read_result.content
for hunk in op.hunks:
search_lines = [l.content for l in hunk.lines if l.prefix in (' ', '-')]
search_lines = [l.content for l in hunk.lines if l.prefix in {' ', '-'}]
if not search_lines:
# Addition-only hunk: validate context hint uniqueness
if hunk.context_hint:
@ -282,7 +282,7 @@ def _validate_operations(
continue
search_pattern = '\n'.join(search_lines)
replace_lines = [l.content for l in hunk.lines if l.prefix in (' ', '+')]
replace_lines = [l.content for l in hunk.lines if l.prefix in {' ', '+'}]
replacement = '\n'.join(replace_lines)
new_simulated, count, _strategy, match_error = fuzzy_find_and_replace(

4
tools/process_registry.py
View file

@ -1237,7 +1237,7 @@ class ProcessRegistry:
killed = 0
for session in targets:
result = self.kill_process(session.id)
if result.get("status") in ("killed", "already_exited"):
if result.get("status") in {"killed", "already_exited"}:
killed += 1
return killed
@ -1446,7 +1446,7 @@ def _handle_process(args, **kw):
if action == "list":
return json.dumps({"processes": process_registry.list_sessions(task_id=task_id)}, ensure_ascii=False)
elif action in ("poll", "log", "wait", "kill", "write", "submit", "close"):
elif action in {"poll", "log", "wait", "kill", "write", "submit", "close"}:
if not session_id:
return tool_error(f"session_id is required for {action}")
if action == "poll":

2
tools/rl_training_tool.py
View file

@ -919,7 +919,7 @@ async def rl_stop_training(run_id: str) -> str:
run_state = _active_runs[run_id]
if run_state.status not in ("running", "starting"):
if run_state.status not in {"running", "starting"}:
return json.dumps({
"message": f"Run '{run_id}' is not running (status: {run_state.status})",
}, indent=2)

20
tools/send_message_tool.py
View file

@ -1034,7 +1034,7 @@ async def _send_discord(token, chat_id, message, thread_id=None, media_files=Non
filename=os.path.basename(media_path),
)
async with session.post(thread_url, headers=auth_headers, data=form, **_req_kw) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
return _error(f"Discord forum thread creation error ({resp.status}): {body}")
data = await resp.json()
@ -1052,7 +1052,7 @@ async def _send_discord(token, chat_id, message, thread_id=None, media_files=Non
},
**_req_kw,
) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
return _error(f"Discord forum thread creation error ({resp.status}): {body}")
data = await resp.json()
@ -1076,7 +1076,7 @@ async def _send_discord(token, chat_id, message, thread_id=None, media_files=Non
# Send text message (skip if empty and media is present)
if message.strip() or not media_files:
async with session.post(url, headers=json_headers, json={"content": message}, **_req_kw) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
return _error(f"Discord API error ({resp.status}): {body}")
last_data = await resp.json()
@ -1094,7 +1094,7 @@ async def _send_discord(token, chat_id, message, thread_id=None, media_files=Non
with open(media_path, "rb") as f:
form.add_field("files[0]", f, filename=filename)
async with session.post(url, headers=auth_headers, data=form, **_req_kw) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
warning = _sanitize_error_text(f"Failed to send media {media_path}: Discord API error ({resp.status}): {body}")
logger.error(warning)
@ -1457,7 +1457,7 @@ async def _send_mattermost(token, extra, chat_id, message):
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
async with session.post(url, headers=headers, json={"channel_id": chat_id, "message": message}) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
return _error(f"Mattermost API error ({resp.status}): {body}")
data = await resp.json()
@ -1501,7 +1501,7 @@ async def _send_matrix(token, extra, chat_id, message):
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
async with session.put(url, headers=headers, json=payload) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
return _error(f"Matrix API error ({resp.status}): {body}")
data = await resp.json()
@ -1585,7 +1585,7 @@ async def _send_homeassistant(token, extra, chat_id, message):
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
async with session.post(url, headers=headers, json={"message": message, "target": chat_id}) as resp:
if resp.status not in (200, 201):
if resp.status not in {200, 201}:
body = await resp.text()
return _error(f"Home Assistant API error ({resp.status}): {body}")
return {"success": True, "platform": "homeassistant", "chat_id": chat_id}
@ -1827,7 +1827,7 @@ async def _send_qqbot(pconfig, chat_id, message):
# Try channel endpoint first (works for guild channels)
url = f"https://api.sgroup.qq.com/channels/{chat_id}/messages"
resp = await client.post(url, json=payload, headers=headers)
if resp.status_code in (200, 201):
if resp.status_code in {200, 201}:
data = resp.json()
return {"success": True, "platform": "qqbot", "chat_id": chat_id,
"message_id": data.get("id")}
@ -1835,7 +1835,7 @@ async def _send_qqbot(pconfig, chat_id, message):
# If channel endpoint failed (likely "频道不存在"), try C2C endpoint
url_c2c = f"https://api.sgroup.qq.com/v2/users/{chat_id}/messages"
resp_c2c = await client.post(url_c2c, json=payload, headers=headers)
if resp_c2c.status_code in (200, 201):
if resp_c2c.status_code in {200, 201}:
data = resp_c2c.json()
return {"success": True, "platform": "qqbot", "chat_id": chat_id,
"message_id": data.get("id")}
@ -1843,7 +1843,7 @@ async def _send_qqbot(pconfig, chat_id, message):
# If C2C also failed, try group endpoint
url_group = f"https://api.sgroup.qq.com/v2/groups/{chat_id}/messages"
resp_group = await client.post(url_group, json=payload, headers=headers)
if resp_group.status_code in (200, 201):
if resp_group.status_code in {200, 201}:
data = resp_group.json()
return {"success": True, "platform": "qqbot", "chat_id": chat_id,
"message_id": data.get("id")}

2
tools/skill_manager_tool.py
View file

@ -780,7 +780,7 @@ def skill_manage(
if action == "create":
if is_background_review():
mark_agent_created(name)
elif action in ("patch", "edit", "write_file", "remove_file"):
elif action in {"patch", "edit", "write_file", "remove_file"}:
bump_patch(name)
elif action == "delete":
forget(name)

2
tools/skills_guard.py
View file

@ -814,7 +814,7 @@ def _check_structure(skill_dir: Path) -> List[Finding]:
))
# Executable permission on non-script files
if ext not in ('.sh', '.bash', '.py', '.rb', '.pl') and f.stat().st_mode & 0o111:
if ext not in {'.sh', '.bash', '.py', '.rb', '.pl'} and f.stat().st_mode & 0o111:
findings.append(Finding(
pattern_id="unexpected_executable",
severity="medium",

4
tools/skills_hub.py
View file

@ -101,7 +101,7 @@ def _normalize_bundle_path(path_value: str, *, field_name: str, allow_nested: bo
normalized = raw.replace("\\", "/")
path = PurePosixPath(normalized)
parts = [part for part in path.parts if part not in ("", ".")]
parts = [part for part in path.parts if part not in {"", "."}]
if normalized.startswith("/") or path.is_absolute():
raise ValueError(f"Unsafe {field_name}: {path_value}")
@ -1415,7 +1415,7 @@ class SkillsShSource(SkillSource):
dir_name = entry["name"]
if dir_name.startswith((".", "_")):
continue
if dir_name in ("skills", ".agents", ".claude"):
if dir_name in {"skills", ".agents", ".claude"}:
continue # already tried
# Try direct: repo/dir/skill_token
direct_id = f"{repo}/{dir_name}/{skill_token}"

4
tools/skills_tool.py
View file

@ -1133,7 +1133,7 @@ def skill_view(
available_files["assets"].append(rel)
elif rel.startswith("scripts/"):
available_files["scripts"].append(rel)
elif f.suffix in [
elif f.suffix in {
".md",
".py",
".yaml",
@ -1141,7 +1141,7 @@ def skill_view(
".json",
".tex",
".sh",
]:
}:
available_files["other"].append(rel)
# Remove empty categories

22
tools/terminal_tool.py
View file

@ -139,7 +139,7 @@ def _check_vercel_sandbox_requirements(config: dict[str, Any]) -> bool:
return False
disk = config.get("container_disk", 51200)
if disk not in (0, 51200):
if disk not in {0, 51200}:
logger.error(
"Vercel Sandbox does not support custom TERMINAL_CONTAINER_DISK=%s. "
"Use the default shared setting (51200 MB).",
@ -416,7 +416,7 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str:
chars = []
while True:
c = msvcrt.getwch()
if c in ("\r", "\n"):
if c in {"\r", "\n"}:
break
if c == "\x03":
raise KeyboardInterrupt
@ -432,7 +432,7 @@ def _prompt_for_sudo_password(timeout_seconds: int = 45) -> str:
chars = []
while True:
b = os.read(tty_fd, 1)
if not b or b in (b"\n", b"\r"):
if not b or b in {b"\n", b"\r"}:
break
chars.append(b)
result["password"] = b"".join(chars).decode("utf-8", errors="replace")
@ -707,7 +707,7 @@ def _rewrite_compound_background(command: str) -> str:
continue
# Quoted tokens — consume whole string via the shared tokenizer.
if ch in ("'", '"'):
if ch in {"'", '"'}:
_, next_i = _read_shell_token(command, i)
i = max(next_i, i + 1)
continue
@ -1009,7 +1009,7 @@ def _get_env_config() -> Dict[str, Any]:
default_image = "nikolaik/python-nodejs:python3.11-nodejs20"
env_type = os.getenv("TERMINAL_ENV", "local")
mount_docker_cwd = os.getenv("TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false").lower() in ("true", "1", "yes")
mount_docker_cwd = os.getenv("TERMINAL_DOCKER_MOUNT_CWD_TO_WORKSPACE", "false").lower() in {"true", "1", "yes"}
# Default cwd: local uses the host's current directory, ssh uses the
# remote home, Vercel uses its documented workspace root, and everything
@ -1041,7 +1041,7 @@ def _get_env_config() -> Dict[str, Any]:
):
host_cwd = candidate
cwd = "/workspace"
elif env_type in ("modal", "docker", "singularity", "daytona", "vercel_sandbox") and cwd:
elif env_type in {"modal", "docker", "singularity", "daytona", "vercel_sandbox"} and cwd:
# Host paths and relative paths that won't work inside containers
is_host_path = any(cwd.startswith(p) for p in host_prefixes)
is_relative = not os.path.isabs(cwd) # e.g. "." or "src/"
@ -1076,17 +1076,17 @@ def _get_env_config() -> Dict[str, Any]:
"ssh_persistent": os.getenv(
"TERMINAL_SSH_PERSISTENT",
os.getenv("TERMINAL_PERSISTENT_SHELL", "true"),
).lower() in ("true", "1", "yes"),
"local_persistent": os.getenv("TERMINAL_LOCAL_PERSISTENT", "false").lower() in ("true", "1", "yes"),
).lower() in {"true", "1", "yes"},
"local_persistent": os.getenv("TERMINAL_LOCAL_PERSISTENT", "false").lower() in {"true", "1", "yes"},
# Container resource config (applies to docker, singularity, modal,
# daytona, and vercel_sandbox -- ignored for local/ssh)
"container_cpu": _parse_env_var("TERMINAL_CONTAINER_CPU", "1", float, "number"),
"container_memory": _parse_env_var("TERMINAL_CONTAINER_MEMORY", "5120"), # MB (default 5GB)
"container_disk": _parse_env_var("TERMINAL_CONTAINER_DISK", "51200"), # MB (default 50GB)
"container_persistent": os.getenv("TERMINAL_CONTAINER_PERSISTENT", "true").lower() in ("true", "1", "yes"),
"container_persistent": os.getenv("TERMINAL_CONTAINER_PERSISTENT", "true").lower() in {"true", "1", "yes"},
"docker_volumes": _parse_env_var("TERMINAL_DOCKER_VOLUMES", "[]", json.loads, "valid JSON"),
"docker_env": _parse_env_var("TERMINAL_DOCKER_ENV", "{}", json.loads, "valid JSON"),
"docker_run_as_host_user": os.getenv("TERMINAL_DOCKER_RUN_AS_HOST_USER", "false").lower() in ("true", "1", "yes"),
"docker_run_as_host_user": os.getenv("TERMINAL_DOCKER_RUN_AS_HOST_USER", "false").lower() in {"true", "1", "yes"},
"docker_extra_args": _parse_env_var("TERMINAL_DOCKER_EXTRA_ARGS", "[]", json.loads, "valid JSON"),
}
@ -1782,7 +1782,7 @@ def terminal_tool(
}
container_config = None
if env_type in ("docker", "singularity", "modal", "daytona", "vercel_sandbox"):
if env_type in {"docker", "singularity", "modal", "daytona", "vercel_sandbox"}:
container_config = {
"container_cpu": config.get("container_cpu", 1),
"container_memory": config.get("container_memory", 5120),

8
tools/tirith_security.py
View file

@ -52,7 +52,7 @@ def _env_bool(key: str, default: bool) -> bool:
val = os.getenv(key)
if val is None:
return default
return val.lower() in ("1", "true", "yes")
return val.lower() in {"1", "true", "yes"}
def _env_int(key: str, default: int) -> int:
@ -189,14 +189,14 @@ def _detect_target() -> str | None:
# Android (Termux) is ABI-compatible with Linux — reuse Linux binaries.
if system == "Darwin":
plat = "apple-darwin"
elif system in ("Linux", "Android"):
elif system in {"Linux", "Android"}:
plat = "unknown-linux-gnu"
else:
return None
if machine in ("x86_64", "amd64"):
if machine in {"x86_64", "amd64"}:
arch = "x86_64"
elif machine in ("aarch64", "arm64"):
elif machine in {"aarch64", "arm64"}:
arch = "aarch64"
else:
return None

2
tools/todo_tool.py
View file

@ -109,7 +109,7 @@ class TodoStore:
# cause the model to re-do finished work after compression.
active_items = [
item for item in self._items
if item["status"] in ("pending", "in_progress")
if item["status"] in {"pending", "in_progress"}
]
if not active_items:
return None

6
tools/tts_tool.py
View file

@ -1612,7 +1612,7 @@ def text_to_speech_tool(
file_path = out_dir / f"tts_{timestamp}.{fmt}"
# Use .ogg for Telegram with providers that support native Opus output,
# otherwise fall back to .mp3 (Edge TTS will attempt ffmpeg conversion later).
elif want_opus and provider in ("openai", "elevenlabs", "mistral", "gemini"):
elif want_opus and provider in {"openai", "elevenlabs", "mistral", "gemini"}:
file_path = out_dir / f"tts_{timestamp}.ogg"
else:
file_path = out_dir / f"tts_{timestamp}.mp3"
@ -1762,12 +1762,12 @@ def text_to_speech_tool(
if opus_path:
file_str = opus_path
voice_compatible = file_str.endswith(".ogg")
elif provider in ("edge", "neutts", "minimax", "xai", "kittentts", "piper") and not file_str.endswith(".ogg"):
elif provider in {"edge", "neutts", "minimax", "xai", "kittentts", "piper"} and not file_str.endswith(".ogg"):
opus_path = _convert_to_opus(file_str)
if opus_path:
file_str = opus_path
voice_compatible = True
elif provider in ("elevenlabs", "openai", "mistral", "gemini"):
elif provider in {"elevenlabs", "openai", "mistral", "gemini"}:
voice_compatible = file_str.endswith(".ogg")
file_size = os.path.getsize(file_str)

4
tools/url_safety.py
View file

@ -96,10 +96,10 @@ def _global_allow_private_urls() -> bool:
# 1. Env var override (highest priority)
env_val = os.getenv("HERMES_ALLOW_PRIVATE_URLS", "").strip().lower()
if env_val in ("true", "1", "yes"):
if env_val in {"true", "1", "yes"}:
_cached_allow_private = True
return _cached_allow_private
if env_val in ("false", "0", "no"):
if env_val in {"false", "0", "no"}:
# Explicit false — don't fall through to config
return _cached_allow_private

2
tools/vision_tools.py
View file

@ -346,7 +346,7 @@ def _resize_image_for_vision(image_path: Path, mime_type: Optional[str] = None,
data_url = _image_to_base64_data_url(image_path, mime_type=mime_type)
return data_url # fall through to size-check in caller
# Convert RGBA to RGB for JPEG output
if pil_format == "JPEG" and img.mode in ("RGBA", "P"):
if pil_format == "JPEG" and img.mode in {"RGBA", "P"}:
img = img.convert("RGB")
# Strategy: halve dimensions until base64 fits, up to 4 rounds.

10
tools/web_tools.py
View file

@ -126,7 +126,7 @@ def _get_backend() -> str:
keys manually without running setup.
"""
configured = (_load_web_config().get("backend") or "").lower().strip()
if configured in ("parallel", "firecrawl", "tavily", "exa", "searxng", "brave-free", "ddgs"):
if configured in {"parallel", "firecrawl", "tavily", "exa", "searxng", "brave-free", "ddgs"}:
return configured
# Fallback for manual / legacy config — pick the highest-priority
@ -1074,7 +1074,7 @@ def _parallel_search(query: str, limit: int = 5) -> dict:
return {"error": "Interrupted", "success": False}
mode = os.getenv("PARALLEL_SEARCH_MODE", "agentic").lower().strip()
if mode not in ("fast", "one-shot", "agentic"):
if mode not in {"fast", "one-shot", "agentic"}:
mode = "agentic"
logger.info("Parallel search: '%s' (mode=%s, limit=%d)", query, mode, limit)
@ -1397,7 +1397,7 @@ async def web_extract_tool(
"include_images": False,
})
results = _normalize_tavily_documents(raw, fallback_url=safe_urls[0] if safe_urls else "")
elif backend in ("searxng", "brave-free", "ddgs"):
elif backend in {"searxng", "brave-free", "ddgs"}:
# These backends are search-only — they cannot extract URL content
_label = {"searxng": "SearXNG", "brave-free": "Brave Search (free tier)", "ddgs": "DuckDuckGo (ddgs)"}[backend]
return json.dumps({
@ -1781,7 +1781,7 @@ async def web_crawl_tool(
return cleaned_result
# SearXNG / Brave Search (free tier) / DuckDuckGo (ddgs) are search-only — they cannot crawl
if backend in ("searxng", "brave-free", "ddgs"):
if backend in {"searxng", "brave-free", "ddgs"}:
_label = {"searxng": "SearXNG", "brave-free": "Brave Search (free tier)", "ddgs": "DuckDuckGo (ddgs)"}[backend]
return json.dumps({
"error": f"{_label} is a search-only backend and cannot crawl URLs. "
@ -2084,7 +2084,7 @@ def check_firecrawl_api_key() -> bool:
def check_web_api_key() -> bool:
"""Check whether the configured web backend is available."""
configured = _load_web_config().get("backend", "").lower().strip()
if configured in ("exa", "parallel", "firecrawl", "tavily", "searxng", "brave-free", "ddgs"):
if configured in {"exa", "parallel", "firecrawl", "tavily", "searxng", "brave-free", "ddgs"}:
return _is_backend_available(configured)
return any(
_is_backend_available(backend)

2
tools/yuanbao_tools.py
View file

@ -122,7 +122,7 @@ async def query_group_members(
hint = {"mention_hint": MENTION_HINT} if mention else {}
if action == "list_bots":
bots = [m for m in all_members if m["role"] in ("yuanbao_ai", "bot")]
bots = [m for m in all_members if m["role"] in {"yuanbao_ai", "bot"}]
if not bots:
return {"success": False, "error": "No bots found in this group."}
return {