| 1 | from __future__ import annotations |
| 2 | |
| 3 | import inspect |
| 4 | import json |
| 5 | import os |
| 6 | import re |
| 7 | import runpy |
| 8 | import shlex |
| 9 | from pathlib import Path |
| 10 | from typing import Any |
| 11 | |
| 12 | import yaml |
| 13 | |
| 14 | from agent import AgentContext |
| 15 | from helpers import files, plugins, projects, yaml as yaml_helper |
| 16 | from helpers.skills import split_frontmatter |
| 17 | |
| 18 | |
| 19 | PLUGIN_NAME = "_commands" |
| 20 | COMMANDS_DIR = "commands" |
| 21 | COMMAND_CONFIG_SUFFIX = ".command.yaml" |
| 22 | LEGACY_COMMAND_FILE_SUFFIX = ".command.md" |
| 23 | TEXT_TEMPLATE_SUFFIX = ".txt" |
| 24 | SCRIPT_TEMPLATE_SUFFIX = ".py" |
| 25 | STANDARD_CONFIG_KEYS = { |
| 26 | "name", |
| 27 | "description", |
| 28 | "argument_hint", |
| 29 | "type", |
| 30 | "template_path", |
| 31 | "script_path", |
| 32 | "include_history", |
| 33 | } |
| 34 | _INVALID_COMMAND_CHARS_RE = re.compile(r"[^a-z0-9_-]+") |
| 35 | _MULTI_DASH_RE = re.compile(r"-{2,}") |
| 36 | _PLACEHOLDER_RE = re.compile(r"\{([a-zA-Z0-9_.-]+)\}") |
| 37 | |
| 38 | |
| 39 | def sanitize_command_name(raw_name: str) -> str: |
| 40 | """Sanitize a raw command name to a lowercase, hyphen-separated slug. |
| 41 | |
| 42 | Strips whitespace, lowercases, replaces spaces and invalid characters with hyphens, |
| 43 | collapses consecutive hyphens, and strips leading/trailing hyphens and underscores. |
| 44 | |
| 45 | Raises: |
| 46 | ValueError: If the resulting name is empty. |
| 47 | |
| 48 | """ |
| 49 | raw = (raw_name or "").strip().lower() |
| 50 | name = raw.replace(" ", "-") |
| 51 | name = _INVALID_COMMAND_CHARS_RE.sub("-", name) |
| 52 | name = _MULTI_DASH_RE.sub("-", name).strip("-_") |
| 53 | if not name: |
| 54 | raise ValueError("Command name must contain at least one letter or number") |
| 55 | return name |
| 56 | |
| 57 | |
| 58 | def normalize_command_type(raw_type: str) -> str: |
| 59 | """Normalize a command type string to either ``"text"`` or ``"script"``.""" |
| 60 | command_type = (raw_type or "text").strip().lower() |
| 61 | if command_type not in {"text", "script"}: |
| 62 | raise ValueError('Command type must be either "text" or "script"') |
| 63 | return command_type |
| 64 | |
| 65 | |
| 66 | def command_file_name(command_name: str) -> str: |
| 67 | """Return the ``.command.yaml`` config filename for *command_name*.""" |
| 68 | return f"{sanitize_command_name(command_name)}{COMMAND_CONFIG_SUFFIX}" |
| 69 | |
| 70 | |
| 71 | def command_content_file_name(command_name: str, command_type: str) -> str: |
| 72 | """Return the content filename (``.txt`` or ``.py``) for *command_name* and *command_type*.""" |
| 73 | suffix = ( |
| 74 | TEXT_TEMPLATE_SUFFIX |
| 75 | if normalize_command_type(command_type) == "text" |
| 76 | else SCRIPT_TEMPLATE_SUFFIX |
| 77 | ) |
| 78 | return f"{sanitize_command_name(command_name)}{suffix}" |
| 79 | |
| 80 | |
| 81 | def parse_slash_invocation(raw_message: str, *, fallback_command: str = "") -> dict[str, Any]: |
| 82 | """Parse a raw slash-command message into its component parts. |
| 83 | |
| 84 | Returns a dict with keys: ``raw_text``, ``command_name``, ``raw_arguments``, |
| 85 | and ``arguments`` (parsed by :func:`parse_arguments`). |
| 86 | |
| 87 | Args: |
| 88 | raw_message: The full message string, e.g. ``"/scan --url https://example.com"``. |
| 89 | fallback_command: Command name to use when no slash prefix is found. |
| 90 | |
| 91 | """ |
| 92 | text = (raw_message or "").strip() |
| 93 | slash_match = re.match(r"^/([^\s]+)(?:\s+([\s\S]*))?$", text) |
| 94 | postfix_match = ( |
| 95 | None |
| 96 | if slash_match |
| 97 | else re.match(r"^([\s\S]*\S)\s+/([^\s]+)\s*$", text) |
| 98 | ) |
| 99 | if slash_match: |
| 100 | try: |
| 101 | command_name = sanitize_command_name(slash_match.group(1)) |
| 102 | except ValueError: |
| 103 | command_name = sanitize_command_name(fallback_command) if fallback_command else "" |
| 104 | raw_arguments = (slash_match.group(2) or "").strip() |
| 105 | elif postfix_match: |
| 106 | try: |
| 107 | command_name = sanitize_command_name(postfix_match.group(2)) |
| 108 | except ValueError: |
| 109 | command_name = sanitize_command_name(fallback_command) if fallback_command else "" |
| 110 | raw_arguments = postfix_match.group(1).strip() |
| 111 | else: |
| 112 | command_name = sanitize_command_name(fallback_command) if fallback_command else "" |
| 113 | raw_arguments = text |
| 114 | |
| 115 | parsed_arguments = parse_arguments(raw_arguments) |
| 116 | return { |
| 117 | "raw_text": text, |
| 118 | "command_name": command_name, |
| 119 | "raw_arguments": raw_arguments, |
| 120 | "arguments": parsed_arguments, |
| 121 | } |
| 122 | |
| 123 | |
| 124 | def parse_arguments(raw_arguments: str) -> dict[str, Any]: |
| 125 | """Parse a raw argument string into positional args, flags, and tokens. |
| 126 | |
| 127 | Supports positional values, long flags (``--key value``, ``--key=value``), |
| 128 | short flags (``-f``), and short flag bundles (``-vq``). |
| 129 | |
| 130 | Returns a dict with keys: ``raw``, ``tokens``, ``positional``, ``flags``. |
| 131 | |
| 132 | """ |
| 133 | normalized_arguments = (raw_arguments or "").strip() |
| 134 | tokens = _split_arguments(normalized_arguments) |
| 135 | positional: list[str] = [] |
| 136 | flags: dict[str, Any] = {} |
| 137 | |
| 138 | index = 0 |
| 139 | while index < len(tokens): |
| 140 | token = tokens[index] |
| 141 | if token.startswith("--") and len(token) > 2: |
| 142 | key, value, consumed = _parse_long_flag(token, tokens, index) |
| 143 | _set_flag_value(flags, key, value) |
| 144 | index += consumed |
| 145 | continue |
| 146 | |
| 147 | if token.startswith("-") and len(token) > 1: |
| 148 | consumed = _parse_short_flag_bundle(token, tokens, index, flags) |
| 149 | index += consumed |
| 150 | continue |
| 151 | |
| 152 | positional.append(token) |
| 153 | index += 1 |
| 154 | |
| 155 | return { |
| 156 | "raw": normalized_arguments, |
| 157 | "tokens": tokens, |
| 158 | "positional": positional, |
| 159 | "flags": flags, |
| 160 | } |
| 161 | |
| 162 | |
| 163 | def render_command_body( |
| 164 | body: str, |
| 165 | raw_arguments: str, |
| 166 | *, |
| 167 | command_name: str = "", |
| 168 | raw_message: str = "", |
| 169 | ) -> str: |
| 170 | """Render *body* as a text template substituting placeholders from *raw_arguments*. |
| 171 | |
| 172 | Args: |
| 173 | body: Template string with ``{placeholder}`` markers. |
| 174 | raw_arguments: Unparsed argument string from the command invocation. |
| 175 | command_name: Optional command name used for slash-invocation parsing fallback. |
| 176 | raw_message: Full original message; when provided takes precedence over raw_arguments. |
| 177 | |
| 178 | """ |
| 179 | invocation = parse_slash_invocation( |
| 180 | raw_message or raw_arguments, |
| 181 | fallback_command=command_name, |
| 182 | ) |
| 183 | if not raw_message: |
| 184 | invocation["raw_arguments"] = (raw_arguments or "").strip() |
| 185 | invocation["arguments"] = parse_arguments(invocation["raw_arguments"]) |
| 186 | return render_text_template(body, invocation) |
| 187 | |
| 188 | |
| 189 | def render_text_template(body: str, invocation: dict[str, Any]) -> str: |
| 190 | """Render *body* as a template substituting placeholders from *invocation* context. |
| 191 | |
| 192 | Unrecognised placeholders resolve to empty string. If *raw_arguments* is present |
| 193 | and the template contains no argument references, the arguments are appended. |
| 194 | |
| 195 | """ |
| 196 | template = body or "" |
| 197 | rendered = template |
| 198 | |
| 199 | context = _build_template_context(invocation) |
| 200 | rendered = _PLACEHOLDER_RE.sub( |
| 201 | lambda match: _resolve_placeholder(match.group(1), context), |
| 202 | rendered, |
| 203 | ) |
| 204 | rendered = _render_legacy_placeholders(rendered, invocation) |
| 205 | rendered = rendered.strip() |
| 206 | |
| 207 | raw_arguments = invocation["raw_arguments"] |
| 208 | if raw_arguments and not _template_references_arguments(template): |
| 209 | suffix = f"Arguments:\n{raw_arguments}" |
| 210 | rendered = f"{rendered}\n\n{suffix}" if rendered else suffix |
| 211 | |
| 212 | return rendered.strip() |
| 213 | |
| 214 | |
| 215 | def get_scope_key(project_name: str = "", agent_profile: str = "") -> str: |
| 216 | """Return the scope identifier key: ``'project'`` when a project is active, else ``'global'``.""" |
| 217 | if project_name: |
| 218 | return "project" |
| 219 | return "global" |
| 220 | |
| 221 | |
| 222 | def get_scope_label(project_name: str = "", agent_profile: str = "") -> str: |
| 223 | """Return the human-readable scope label: ``'Project'`` or ``'Global'``.""" |
| 224 | if project_name: |
| 225 | return "Project" |
| 226 | return "Global" |
| 227 | |
| 228 | |
| 229 | def get_scope_directory(project_name: str = "", agent_profile: str = "") -> str: |
| 230 | """Return the absolute filesystem path to the commands directory for the given scope.""" |
| 231 | return plugins.determine_plugin_asset_path( |
| 232 | PLUGIN_NAME, |
| 233 | project_name, |
| 234 | "", |
| 235 | COMMANDS_DIR, |
| 236 | ) |
| 237 | |
| 238 | |
| 239 | def ensure_scope_directory(project_name: str = "", agent_profile: str = "") -> str: |
| 240 | """Ensure the commands directory for the given scope exists, creating it when absent. |
| 241 | |
| 242 | Returns the absolute path to the directory. |
| 243 | |
| 244 | """ |
| 245 | directory = get_scope_directory(project_name, "") |
| 246 | Path(directory).mkdir(parents=True, exist_ok=True) |
| 247 | return directory |
| 248 | |
| 249 | |
| 250 | def get_scope_payload( |
| 251 | project_name: str = "", |
| 252 | agent_profile: str = "", |
| 253 | *, |
| 254 | ensure_directory: bool = False, |
| 255 | ) -> dict[str, Any]: |
| 256 | """Build a scope descriptor dict for the given project/agent context. |
| 257 | |
| 258 | Returns a dict containing ``project_name``, ``scope_key``, ``scope_label``, |
| 259 | ``directory_path``, ``exists``, and the private ``_directory_abs_path`` key. |
| 260 | |
| 261 | Args: |
| 262 | ensure_directory: When ``True``, create the directory if it does not exist. |
| 263 | |
| 264 | """ |
| 265 | directory_path = ( |
| 266 | ensure_scope_directory(project_name, "") |
| 267 | if ensure_directory |
| 268 | else get_scope_directory(project_name, "") |
| 269 | ) |
| 270 | return { |
| 271 | "project_name": project_name, |
| 272 | "scope_key": get_scope_key(project_name, ""), |
| 273 | "scope_label": get_scope_label(project_name, ""), |
| 274 | "directory_path": _normalize_client_path(directory_path), |
| 275 | "exists": os.path.isdir(directory_path), |
| 276 | "_directory_abs_path": directory_path, |
| 277 | } |
| 278 | |
| 279 | |
| 280 | def get_context_scope(context_id: str = "") -> dict[str, str]: |
| 281 | """Resolve the active project name for *context_id* and return a scope mapping. |
| 282 | |
| 283 | Returns ``{"project_name": str}`` — empty string when no project is associated. |
| 284 | |
| 285 | """ |
| 286 | context = _get_context(context_id) |
| 287 | if not context: |
| 288 | return {"project_name": ""} |
| 289 | |
| 290 | return { |
| 291 | "project_name": projects.get_context_project_name(context) or "", |
| 292 | } |
| 293 | |
| 294 | |
| 295 | def list_scope_commands( |
| 296 | project_name: str = "", |
| 297 | agent_profile: str = "", |
| 298 | ) -> tuple[list[dict[str, Any]], dict[str, Any]]: |
| 299 | """List all commands defined in *project_name* scope (not merged with global). |
| 300 | |
| 301 | Each command entry includes ``override_scopes`` and ``override_count`` fields |
| 302 | indicating lower-scoped commands with the same name. |
| 303 | |
| 304 | Returns: |
| 305 | Tuple of (commands list, stripped scope payload dict). |
| 306 | |
| 307 | """ |
| 308 | scope = get_scope_payload(project_name, "") |
| 309 | commands = _load_scope_commands(project_name) |
| 310 | overrides = _collect_lower_scope_matches(project_name) |
| 311 | |
| 312 | for command in commands: |
| 313 | override_scopes = overrides.get(command["name"], []) |
| 314 | command["override_scopes"] = override_scopes |
| 315 | command["override_count"] = len(override_scopes) |
| 316 | |
| 317 | return commands, strip_private_scope(scope) |
| 318 | |
| 319 | |
| 320 | def list_effective_commands( |
| 321 | project_name: str = "", |
| 322 | agent_profile: str = "", |
| 323 | ) -> tuple[list[dict[str, Any]], dict[str, Any]]: |
| 324 | """Return the merged effective command list for *project_name* scope. |
| 325 | |
| 326 | Project-scoped commands take precedence over global commands of the same name. |
| 327 | Commands are sorted alphabetically by name. |
| 328 | |
| 329 | Returns: |
| 330 | Tuple of (sorted commands list, stripped scope payload dict). |
| 331 | |
| 332 | """ |
| 333 | resolved_scope = get_scope_payload(project_name, "") |
| 334 | merged: dict[str, dict[str, Any]] = {} |
| 335 | |
| 336 | for scope_project in _iter_precedence_scopes(project_name): |
| 337 | for command in _load_scope_commands(scope_project): |
| 338 | merged.setdefault(command["name"], command) |
| 339 | |
| 340 | for command in _discover_builtin_commands(): |
| 341 | merged.setdefault(command["name"], command) |
| 342 | |
| 343 | for command in _discover_plugin_commands(): |
| 344 | merged.setdefault(command["name"], command) |
| 345 | |
| 346 | effective = sorted(merged.values(), key=lambda item: item["name"]) |
| 347 | return effective, strip_private_scope(resolved_scope) |
| 348 | |
| 349 | |
| 350 | def list_builtin_commands() -> list[dict[str, Any]]: |
| 351 | """Return bundled commands for display in the command manager.""" |
| 352 | return sorted(_discover_builtin_commands(), key=lambda item: item["name"]) |
| 353 | |
| 354 | |
| 355 | def get_command( |
| 356 | path: str, |
| 357 | project_name: str = "", |
| 358 | agent_profile: str = "", |
| 359 | ) -> dict[str, Any]: |
| 360 | """Load and return a single command by its config file *path*. |
| 361 | |
| 362 | Validates that *path* belongs to an effective scope for *project_name*. |
| 363 | |
| 364 | Raises: |
| 365 | FileNotFoundError: If the command file does not exist. |
| 366 | ValueError: If the path is outside all valid scopes, not a recognised |
| 367 | config suffix, or the file content is invalid. |
| 368 | |
| 369 | """ |
| 370 | command_path = _validate_command_path(path, project_name, "", allow_plugin=True) |
| 371 | # Determine actual scope from the resolved path to get correct metadata. |
| 372 | # A global command loaded with a project context must report scope=global. |
| 373 | actual_project = "" |
| 374 | for scope in _iter_precedence_scopes(project_name): |
| 375 | scope_dir = get_scope_directory(scope, "") |
| 376 | if files.is_in_dir(command_path, scope_dir): |
| 377 | actual_project = scope |
| 378 | break |
| 379 | command = _load_command_file(command_path, project_name=actual_project) |
| 380 | if not command: |
| 381 | raise ValueError("Command file is invalid or missing required configuration") |
| 382 | if _is_builtin_command_path(command_path): |
| 383 | return _mark_builtin_command(command) |
| 384 | plugin_name = _plugin_name_for_commands_path(command_path) |
| 385 | if plugin_name: |
| 386 | _mark_plugin_command(command, plugin_name) |
| 387 | return command |
| 388 | |
| 389 | |
| 390 | def save_command( |
| 391 | *, |
| 392 | project_name: str = "", |
| 393 | agent_profile: str = "", |
| 394 | existing_path: str = "", |
| 395 | name: str, |
| 396 | description: str, |
| 397 | argument_hint: str = "", |
| 398 | command_type: str = "text", |
| 399 | body: str = "", |
| 400 | include_history: bool = False, |
| 401 | extra_frontmatter: dict[str, Any] | None = None, |
| 402 | ) -> dict[str, Any]: |
| 403 | """Create or update a command, writing both the config and content files. |
| 404 | |
| 405 | When *existing_path* is provided, the old files are removed after the new |
| 406 | files are written (rename/move semantics). |
| 407 | |
| 408 | Returns: |
| 409 | The fully-loaded command dict for the saved command. |
| 410 | |
| 411 | Raises: |
| 412 | FileExistsError: If a command with the same name already exists in scope. |
| 413 | ValueError: If required fields are missing or values are invalid. |
| 414 | |
| 415 | """ |
| 416 | command_name = sanitize_command_name(name) |
| 417 | command_description = (description or "").strip() |
| 418 | if not command_description: |
| 419 | raise ValueError("Command description is required") |
| 420 | |
| 421 | normalized_type = normalize_command_type(command_type) |
| 422 | scope_dir = ensure_scope_directory(project_name, "") |
| 423 | target_config_path = files.get_abs_path(scope_dir, command_file_name(command_name)) |
| 424 | target_content_name = command_content_file_name(command_name, normalized_type) |
| 425 | target_content_path = files.get_abs_path(scope_dir, target_content_name) |
| 426 | existing_abs_path = "" |
| 427 | existing_command: dict[str, Any] | None = None |
| 428 | if existing_path: |
| 429 | try: |
| 430 | existing_abs_path = _validate_command_path(existing_path, project_name, "") |
| 431 | existing_command = _load_command_file(existing_abs_path, project_name=project_name) |
| 432 | except FileNotFoundError: |
| 433 | existing_abs_path = "" |
| 434 | existing_command = None |
| 435 | |
| 436 | if existing_abs_path and not os.path.exists(existing_abs_path): |
| 437 | existing_abs_path = "" |
| 438 | existing_command = None |
| 439 | |
| 440 | if os.path.exists(target_config_path) and not _paths_equal( |
| 441 | target_config_path, existing_abs_path |
| 442 | ): |
| 443 | raise FileExistsError(f'A command named "{command_name}" already exists in this scope') |
| 444 | |
| 445 | existing_content_path = _to_abs_path(existing_command.get("content_path", "")) if existing_command else "" |
| 446 | if os.path.exists(target_content_path) and not _paths_equal( |
| 447 | target_content_path, existing_content_path |
| 448 | ): |
| 449 | raise FileExistsError( |
| 450 | f'Command content file "{Path(target_content_path).name}" already exists in this scope' |
| 451 | ) |
| 452 | |
| 453 | content_key = "template_path" if normalized_type == "text" else "script_path" |
| 454 | config = _build_command_config( |
| 455 | name=command_name, |
| 456 | description=command_description, |
| 457 | argument_hint=argument_hint, |
| 458 | command_type=normalized_type, |
| 459 | content_path=target_content_name, |
| 460 | include_history=include_history, |
| 461 | extra_config=extra_frontmatter or {}, |
| 462 | ) |
| 463 | if content_key not in config: |
| 464 | config[content_key] = target_content_name |
| 465 | |
| 466 | files.write_file(target_content_path, _normalize_command_body(body, normalized_type)) |
| 467 | files.write_file(target_config_path, _build_command_yaml(config)) |
| 468 | |
| 469 | if existing_abs_path and not _paths_equal(existing_abs_path, target_config_path): |
| 470 | files.delete_file(existing_abs_path) |
| 471 | |
| 472 | if ( |
| 473 | existing_content_path |
| 474 | and os.path.exists(existing_content_path) |
| 475 | and not _paths_equal(existing_content_path, target_content_path) |
| 476 | ): |
| 477 | files.delete_file(existing_content_path) |
| 478 | |
| 479 | return get_command(target_config_path, project_name, "") |
| 480 | |
| 481 | |
| 482 | def delete_command( |
| 483 | path: str, |
| 484 | project_name: str = "", |
| 485 | agent_profile: str = "", |
| 486 | ) -> None: |
| 487 | """Delete the config file and associated content file for *path*. |
| 488 | |
| 489 | Raises: |
| 490 | FileNotFoundError: If the command file does not exist. |
| 491 | ValueError: If *path* is invalid or outside the allowed scope. |
| 492 | |
| 493 | """ |
| 494 | command = get_command(path, project_name, "") |
| 495 | command_path = _validate_command_path(path, project_name, "") |
| 496 | files.delete_file(command_path) |
| 497 | |
| 498 | content_path = _to_abs_path(command.get("content_path", "")) |
| 499 | if content_path and os.path.exists(content_path): |
| 500 | files.delete_file(content_path) |
| 501 | |
| 502 | |
| 503 | def duplicate_command( |
| 504 | path: str, |
| 505 | project_name: str = "", |
| 506 | agent_profile: str = "", |
| 507 | ) -> dict[str, Any]: |
| 508 | """Duplicate a command, preserving built-in names so the copy overrides the default. |
| 509 | |
| 510 | Returns the newly created command dict. |
| 511 | |
| 512 | Raises: |
| 513 | FileNotFoundError: If the source command does not exist. |
| 514 | ValueError: If the source path is invalid. |
| 515 | |
| 516 | """ |
| 517 | command = get_command(path, project_name, "") |
| 518 | duplicated_name = ( |
| 519 | command["name"] |
| 520 | if command.get("scope_key") == "builtin" |
| 521 | else _generate_duplicate_name(command["name"], project_name=project_name) |
| 522 | ) |
| 523 | return save_command( |
| 524 | project_name=project_name, |
| 525 | name=duplicated_name, |
| 526 | description=command["description"], |
| 527 | argument_hint=command.get("argument_hint", ""), |
| 528 | command_type=command.get("command_type", "text"), |
| 529 | body=command.get("body", ""), |
| 530 | include_history=bool(command.get("include_history", False)), |
| 531 | extra_frontmatter=command.get("frontmatter_extra", {}), |
| 532 | ) |
| 533 | |
| 534 | |
| 535 | async def resolve_command_invocation( |
| 536 | *, |
| 537 | path: str, |
| 538 | slash_text: str, |
| 539 | project_name: str = "", |
| 540 | context_id: str = "", |
| 541 | ) -> dict[str, Any]: |
| 542 | """Resolve a slash command invocation, executing text rendering or a Python script hook. |
| 543 | |
| 544 | Args: |
| 545 | path: Path to the command config file. |
| 546 | slash_text: The full slash text entered by the user. |
| 547 | project_name: Active project name (empty string for global scope).' |
| 548 | context_id: Agent context ID, used for script commands that request history. |
| 549 | |
| 550 | Returns: |
| 551 | Dict with keys ``command``, ``invocation``, and ``result`` |
| 552 | (``{"text": str, "effects": list}``). |
| 553 | |
| 554 | Raises: |
| 555 | FileNotFoundError: If the command file is not found. |
| 556 | ValueError: If path or slash_text is invalid. |
| 557 | |
| 558 | """ |
| 559 | command = get_command(path, project_name, "") |
| 560 | invocation = parse_slash_invocation(slash_text, fallback_command=command["name"]) |
| 561 | |
| 562 | if command.get("command_type") == "script": |
| 563 | result = await _run_script_command( |
| 564 | command=command, |
| 565 | invocation=invocation, |
| 566 | project_name=project_name, |
| 567 | context_id=context_id, |
| 568 | ) |
| 569 | else: |
| 570 | text = render_text_template(command.get("body", ""), invocation) |
| 571 | result = {"text": text, "effects": []} |
| 572 | |
| 573 | return { |
| 574 | "command": _public_command_payload(command), |
| 575 | "invocation": invocation, |
| 576 | "result": result, |
| 577 | } |
| 578 | |
| 579 | |
| 580 | async def resolve_message_command( |
| 581 | raw_message: str, |
| 582 | *, |
| 583 | context_id: str = "", |
| 584 | ) -> dict[str, Any] | None: |
| 585 | """Resolve a known prefix or postfix slash command from an incoming message.""" |
| 586 | invocation = parse_slash_invocation(raw_message) |
| 587 | command_name = invocation["command_name"] |
| 588 | if not command_name: |
| 589 | return None |
| 590 | |
| 591 | project_name = get_context_scope(context_id)["project_name"] |
| 592 | effective, _ = list_effective_commands(project_name) |
| 593 | command = next((item for item in effective if item["name"] == command_name), None) |
| 594 | if not command: |
| 595 | return None |
| 596 | |
| 597 | return await resolve_command_invocation( |
| 598 | path=command["path"], |
| 599 | slash_text=raw_message, |
| 600 | project_name=project_name, |
| 601 | context_id=context_id, |
| 602 | ) |
| 603 | |
| 604 | |
| 605 | def _build_command_config( |
| 606 | *, |
| 607 | name: str, |
| 608 | description: str, |
| 609 | argument_hint: str, |
| 610 | command_type: str, |
| 611 | content_path: str, |
| 612 | include_history: bool, |
| 613 | extra_config: dict[str, Any], |
| 614 | ) -> dict[str, Any]: |
| 615 | config: dict[str, Any] = { |
| 616 | "name": name, |
| 617 | "description": description, |
| 618 | "type": command_type, |
| 619 | } |
| 620 | clean_argument_hint = (argument_hint or "").strip() |
| 621 | if clean_argument_hint: |
| 622 | config["argument_hint"] = clean_argument_hint |
| 623 | |
| 624 | if command_type == "text": |
| 625 | config["template_path"] = content_path |
| 626 | else: |
| 627 | config["script_path"] = content_path |
| 628 | if include_history: |
| 629 | config["include_history"] = True |
| 630 | |
| 631 | for key, value in (extra_config or {}).items(): |
| 632 | if key in STANDARD_CONFIG_KEYS: |
| 633 | continue |
| 634 | config[key] = value |
| 635 | |
| 636 | return config |
| 637 | |
| 638 | |
| 639 | def _build_command_yaml(config: dict[str, Any]) -> str: |
| 640 | return f"{yaml_helper.dumps(config).strip()}\n" |
| 641 | |
| 642 | |
| 643 | def _normalize_command_body(body: str, command_type: str) -> str: |
| 644 | cleaned = (body or "").lstrip("\n").rstrip() |
| 645 | if cleaned: |
| 646 | return f"{cleaned}\n" |
| 647 | return "" |
| 648 | |
| 649 | |
| 650 | def _load_command_file( |
| 651 | file_path: str, |
| 652 | *, |
| 653 | project_name: str = "", |
| 654 | ) -> dict[str, Any] | None: |
| 655 | if file_path.endswith(COMMAND_CONFIG_SUFFIX): |
| 656 | return _load_yaml_command_file(file_path, project_name=project_name) |
| 657 | if file_path.endswith(LEGACY_COMMAND_FILE_SUFFIX): |
| 658 | return _load_legacy_markdown_file(file_path, project_name=project_name) |
| 659 | return None |
| 660 | |
| 661 | |
| 662 | def _load_yaml_command_file( |
| 663 | file_path: str, |
| 664 | *, |
| 665 | project_name: str = "", |
| 666 | ) -> dict[str, Any] | None: |
| 667 | try: |
| 668 | raw_content = files.read_file(file_path) |
| 669 | except FileNotFoundError: |
| 670 | return None |
| 671 | |
| 672 | try: |
| 673 | parsed = yaml.safe_load(raw_content) or {} |
| 674 | except yaml.YAMLError: |
| 675 | return None |
| 676 | if not isinstance(parsed, dict): |
| 677 | return None |
| 678 | |
| 679 | raw_name = str(parsed.get("name") or "").strip() |
| 680 | description = str(parsed.get("description") or "").strip() |
| 681 | if not raw_name or not description: |
| 682 | return None |
| 683 | |
| 684 | try: |
| 685 | command_name = sanitize_command_name(raw_name) |
| 686 | command_type = normalize_command_type(str(parsed.get("type") or "text")) |
| 687 | except ValueError: |
| 688 | return None |
| 689 | |
| 690 | directory_path = str(Path(file_path).parent) |
| 691 | content_key = "template_path" if command_type == "text" else "script_path" |
| 692 | configured_content_path = str(parsed.get(content_key) or "").strip() or command_content_file_name( |
| 693 | command_name, command_type |
| 694 | ) |
| 695 | content_abs_path = files.get_abs_path(directory_path, configured_content_path) |
| 696 | # Content file must live in the same directory as its config file. |
| 697 | # Using directory_path (not the project scope root) allows global commands |
| 698 | # to load correctly even when a project context is active. |
| 699 | if not files.is_in_dir(content_abs_path, directory_path): |
| 700 | return None |
| 701 | |
| 702 | try: |
| 703 | body = files.read_file(content_abs_path) |
| 704 | except FileNotFoundError: |
| 705 | body = "" |
| 706 | |
| 707 | argument_hint = str(parsed.get("argument_hint") or "").strip() |
| 708 | include_history = bool(parsed.get("include_history", False)) |
| 709 | extra_config = { |
| 710 | key: value for key, value in parsed.items() if key not in STANDARD_CONFIG_KEYS |
| 711 | } |
| 712 | |
| 713 | return { |
| 714 | "name": command_name, |
| 715 | "description": description, |
| 716 | "argument_hint": argument_hint, |
| 717 | "command_type": command_type, |
| 718 | "include_history": include_history, |
| 719 | "body": body, |
| 720 | "path": _normalize_client_path(file_path), |
| 721 | "config_path": _normalize_client_path(file_path), |
| 722 | "content_path": _normalize_client_path(content_abs_path), |
| 723 | "directory_path": _normalize_client_path(directory_path), |
| 724 | "project_name": project_name, |
| 725 | "scope_key": get_scope_key(project_name, ""), |
| 726 | "scope_label": get_scope_label(project_name, ""), |
| 727 | "source_scope_key": get_scope_key(project_name, ""), |
| 728 | "source_scope_label": get_scope_label(project_name, ""), |
| 729 | "frontmatter_extra": extra_config, |
| 730 | } |
| 731 | |
| 732 | |
| 733 | def _load_legacy_markdown_file( |
| 734 | file_path: str, |
| 735 | *, |
| 736 | project_name: str = "", |
| 737 | ) -> dict[str, Any] | None: |
| 738 | try: |
| 739 | content = files.read_file(file_path) |
| 740 | except FileNotFoundError: |
| 741 | return None |
| 742 | |
| 743 | frontmatter, body, errors = split_frontmatter(content) |
| 744 | if errors: |
| 745 | return None |
| 746 | |
| 747 | raw_name = str(frontmatter.get("name") or "").strip() |
| 748 | description = str(frontmatter.get("description") or "").strip() |
| 749 | if not raw_name or not description: |
| 750 | return None |
| 751 | |
| 752 | try: |
| 753 | command_name = sanitize_command_name(raw_name) |
| 754 | except ValueError: |
| 755 | return None |
| 756 | |
| 757 | argument_hint = str(frontmatter.get("argument_hint") or "").strip() |
| 758 | extra_frontmatter = { |
| 759 | key: value for key, value in frontmatter.items() if key not in {"name", "description", "argument_hint"} |
| 760 | } |
| 761 | directory_path = str(Path(file_path).parent) |
| 762 | return { |
| 763 | "name": command_name, |
| 764 | "description": description, |
| 765 | "argument_hint": argument_hint, |
| 766 | "command_type": "text", |
| 767 | "include_history": False, |
| 768 | "body": body, |
| 769 | "path": _normalize_client_path(file_path), |
| 770 | "config_path": _normalize_client_path(file_path), |
| 771 | "content_path": _normalize_client_path(file_path), |
| 772 | "directory_path": _normalize_client_path(directory_path), |
| 773 | "project_name": project_name, |
| 774 | "scope_key": get_scope_key(project_name, ""), |
| 775 | "scope_label": get_scope_label(project_name, ""), |
| 776 | "source_scope_key": get_scope_key(project_name, ""), |
| 777 | "source_scope_label": get_scope_label(project_name, ""), |
| 778 | "frontmatter_extra": extra_frontmatter, |
| 779 | } |
| 780 | |
| 781 | |
| 782 | def _validate_command_path( |
| 783 | path: str, |
| 784 | project_name: str = "", |
| 785 | agent_profile: str = "", |
| 786 | *, |
| 787 | allow_plugin: bool = False, |
| 788 | ) -> str: |
| 789 | command_path = _to_abs_path(path) |
| 790 | # Allow commands from any effective scope (project overrides global, but global is also valid) |
| 791 | valid_roots = [get_scope_directory(scope, "") for scope in _iter_precedence_scopes(project_name)] |
| 792 | if not any(files.is_in_dir(command_path, scope_root) for scope_root in valid_roots): |
| 793 | is_builtin = _is_builtin_command_path(command_path) |
| 794 | plugin_name = "" if is_builtin else _plugin_name_for_commands_path(command_path) |
| 795 | if plugin_name: |
| 796 | if not allow_plugin: |
| 797 | raise ValueError("Plugin commands are read-only") |
| 798 | elif is_builtin: |
| 799 | if not allow_plugin: |
| 800 | raise ValueError("Built-in commands are read-only") |
| 801 | else: |
| 802 | raise ValueError("Command path is outside the selected scope") |
| 803 | if not ( |
| 804 | command_path.endswith(COMMAND_CONFIG_SUFFIX) |
| 805 | or command_path.endswith(LEGACY_COMMAND_FILE_SUFFIX) |
| 806 | ): |
| 807 | raise ValueError("Command path must point to a .command.yaml or .command.md file") |
| 808 | if not os.path.exists(command_path): |
| 809 | raise FileNotFoundError("Command file not found") |
| 810 | return command_path |
| 811 | |
| 812 | |
| 813 | def _iter_precedence_scopes(project_name: str) -> list[str]: |
| 814 | if project_name: |
| 815 | return [project_name, ""] |
| 816 | return [""] |
| 817 | |
| 818 | |
| 819 | def _list_scope_files(scope_dir: str) -> list[str]: |
| 820 | if not os.path.isdir(scope_dir): |
| 821 | return [] |
| 822 | files_in_scope = [ |
| 823 | str(path) |
| 824 | for suffix in (COMMAND_CONFIG_SUFFIX, LEGACY_COMMAND_FILE_SUFFIX) |
| 825 | for path in Path(scope_dir).glob(f"*{suffix}") |
| 826 | if path.is_file() |
| 827 | ] |
| 828 | files_in_scope.sort(key=lambda item: Path(item).name.lower()) |
| 829 | return files_in_scope |
| 830 | |
| 831 | |
| 832 | def _load_scope_commands(project_name: str = "") -> list[dict[str, Any]]: |
| 833 | commands: list[dict[str, Any]] = [] |
| 834 | scope_dir = get_scope_directory(project_name, "") |
| 835 | |
| 836 | for file_path in _list_scope_files(scope_dir): |
| 837 | command = _load_command_file(file_path, project_name=project_name) |
| 838 | if command: |
| 839 | commands.append(command) |
| 840 | |
| 841 | commands.sort(key=lambda item: item["name"]) |
| 842 | return commands |
| 843 | |
| 844 | |
| 845 | def _discover_plugin_commands() -> list[dict[str, Any]]: |
| 846 | """Discover commands contributed by enabled plugins.""" |
| 847 | commands: list[dict[str, Any]] = [] |
| 848 | for plugin_name in plugins.get_enabled_plugins(None): |
| 849 | if plugin_name == PLUGIN_NAME: |
| 850 | continue |
| 851 | plugin_dir = plugins.find_plugin_dir(plugin_name) |
| 852 | if not plugin_dir: |
| 853 | continue |
| 854 | plugin_commands_dir = files.get_abs_path(plugin_dir, COMMANDS_DIR) |
| 855 | if not os.path.isdir(plugin_commands_dir): |
| 856 | continue |
| 857 | for file_path in _list_scope_files(plugin_commands_dir): |
| 858 | command = _load_command_file(file_path, project_name="") |
| 859 | if command: |
| 860 | commands.append(_mark_plugin_command(command, plugin_name)) |
| 861 | return commands |
| 862 | |
| 863 | |
| 864 | def _discover_builtin_commands() -> list[dict[str, Any]]: |
| 865 | plugin_dir = plugins.find_plugin_dir(PLUGIN_NAME) |
| 866 | if not plugin_dir: |
| 867 | return [] |
| 868 | commands_dir = files.get_abs_path(plugin_dir, COMMANDS_DIR) |
| 869 | commands: list[dict[str, Any]] = [] |
| 870 | for file_path in _list_scope_files(commands_dir): |
| 871 | command = _load_command_file(file_path, project_name="") |
| 872 | if command: |
| 873 | commands.append(_mark_builtin_command(command)) |
| 874 | return commands |
| 875 | |
| 876 | |
| 877 | def _collect_lower_scope_matches(project_name: str = "") -> dict[str, list[str]]: |
| 878 | lower_scope_matches: dict[str, list[str]] = {} |
| 879 | if not project_name: |
| 880 | return lower_scope_matches |
| 881 | |
| 882 | for command in _load_scope_commands(""): |
| 883 | lower_scope_matches.setdefault(command["name"], []).append(get_scope_label("", "")) |
| 884 | |
| 885 | return lower_scope_matches |
| 886 | |
| 887 | |
| 888 | def _generate_duplicate_name( |
| 889 | command_name: str, |
| 890 | *, |
| 891 | project_name: str = "", |
| 892 | ) -> str: |
| 893 | base_name = sanitize_command_name(f"{command_name}-copy") |
| 894 | candidate = base_name |
| 895 | counter = 2 |
| 896 | scope_dir = ensure_scope_directory(project_name, "") |
| 897 | |
| 898 | while os.path.exists(files.get_abs_path(scope_dir, command_file_name(candidate))): |
| 899 | candidate = f"{base_name}-{counter}" |
| 900 | counter += 1 |
| 901 | |
| 902 | return candidate |
| 903 | |
| 904 | |
| 905 | def _build_template_context(invocation: dict[str, Any]) -> dict[str, Any]: |
| 906 | arguments = invocation.get("arguments", {}) |
| 907 | return { |
| 908 | "full": invocation.get("raw_text", ""), |
| 909 | "raw": invocation.get("raw_arguments", ""), |
| 910 | "command": invocation.get("command_name", ""), |
| 911 | "args": { |
| 912 | "raw": arguments.get("raw", ""), |
| 913 | "tokens": arguments.get("tokens", []), |
| 914 | "positional": arguments.get("positional", []), |
| 915 | "flags": arguments.get("flags", {}), |
| 916 | }, |
| 917 | } |
| 918 | |
| 919 | |
| 920 | def _resolve_placeholder(path: str, context: dict[str, Any]) -> str: |
| 921 | resolved = _resolve_path(context, path) |
| 922 | if resolved is None: |
| 923 | return "" |
| 924 | if isinstance(resolved, (dict, list)): |
| 925 | return json.dumps(resolved, ensure_ascii=False) |
| 926 | return str(resolved) |
| 927 | |
| 928 | |
| 929 | def _resolve_path(value: Any, path: str) -> Any: |
| 930 | current = value |
| 931 | for part in path.split("."): |
| 932 | if isinstance(current, dict): |
| 933 | if part in current: |
| 934 | current = current[part] |
| 935 | continue |
| 936 | part_with_dash = part.replace("_", "-") |
| 937 | if part_with_dash in current: |
| 938 | current = current[part_with_dash] |
| 939 | continue |
| 940 | return None |
| 941 | |
| 942 | if isinstance(current, list): |
| 943 | if not part.isdigit(): |
| 944 | return None |
| 945 | index = int(part) |
| 946 | if index < 0 or index >= len(current): |
| 947 | return None |
| 948 | current = current[index] |
| 949 | continue |
| 950 | |
| 951 | return None |
| 952 | return current |
| 953 | |
| 954 | |
| 955 | def _render_legacy_placeholders(template: str, invocation: dict[str, Any]) -> str: |
| 956 | rendered = template |
| 957 | arguments = invocation.get("arguments", {}) |
| 958 | positional = arguments.get("positional", []) |
| 959 | for index in range(10): |
| 960 | rendered = rendered.replace(f"${index}", positional[index] if index < len(positional) else "") |
| 961 | rendered = rendered.replace("$ARGUMENTS", invocation.get("raw_arguments", "")) |
| 962 | return rendered |
| 963 | |
| 964 | |
| 965 | def _template_references_arguments(template: str) -> bool: |
| 966 | if "$ARGUMENTS" in template: |
| 967 | return True |
| 968 | if any(f"${index}" in template for index in range(10)): |
| 969 | return True |
| 970 | if "{raw}" in template: |
| 971 | return True |
| 972 | return "{args." in template |
| 973 | |
| 974 | |
| 975 | def _parse_long_flag(token: str, tokens: list[str], index: int) -> tuple[str, Any, int]: |
| 976 | flag_token = token[2:] |
| 977 | if "=" in flag_token: |
| 978 | key, value = flag_token.split("=", 1) |
| 979 | return _normalize_flag_name(key), value, 1 |
| 980 | |
| 981 | key = _normalize_flag_name(flag_token) |
| 982 | next_index = index + 1 |
| 983 | if next_index < len(tokens) and not tokens[next_index].startswith("-"): |
| 984 | return key, tokens[next_index], 2 |
| 985 | return key, True, 1 |
| 986 | |
| 987 | |
| 988 | def _parse_short_flag_bundle( |
| 989 | token: str, tokens: list[str], index: int, flags: dict[str, Any] |
| 990 | ) -> int: |
| 991 | short_token = token[1:] |
| 992 | if len(short_token) > 1 and "=" not in short_token: |
| 993 | for char in short_token: |
| 994 | _set_flag_value(flags, _normalize_flag_name(char), True) |
| 995 | return 1 |
| 996 | |
| 997 | if "=" in short_token: |
| 998 | key, value = short_token.split("=", 1) |
| 999 | _set_flag_value(flags, _normalize_flag_name(key), value) |
| 1000 | return 1 |
| 1001 | |
| 1002 | key = _normalize_flag_name(short_token) |
| 1003 | _set_flag_value(flags, key, True) |
| 1004 | return 1 |
| 1005 | |
| 1006 | |
| 1007 | def _set_flag_value(flags: dict[str, Any], key: str, value: Any) -> None: |
| 1008 | if key in flags: |
| 1009 | current = flags[key] |
| 1010 | if isinstance(current, list): |
| 1011 | current.append(value) |
| 1012 | else: |
| 1013 | flags[key] = [current, value] |
| 1014 | return |
| 1015 | flags[key] = value |
| 1016 | |
| 1017 | |
| 1018 | def _normalize_flag_name(raw_flag: str) -> str: |
| 1019 | return (raw_flag or "").strip().lower().replace("-", "_") |
| 1020 | |
| 1021 | |
| 1022 | def _split_arguments(raw_arguments: str) -> list[str]: |
| 1023 | if not raw_arguments: |
| 1024 | return [] |
| 1025 | try: |
| 1026 | return shlex.split(raw_arguments) |
| 1027 | except ValueError: |
| 1028 | return raw_arguments.split() |
| 1029 | |
| 1030 | |
| 1031 | async def _run_script_command( |
| 1032 | *, |
| 1033 | command: dict[str, Any], |
| 1034 | invocation: dict[str, Any], |
| 1035 | project_name: str, |
| 1036 | context_id: str, |
| 1037 | ) -> dict[str, Any]: |
| 1038 | script_path = _to_abs_path(command.get("content_path", "")) |
| 1039 | if not script_path: |
| 1040 | raise ValueError("Script command is missing script_path") |
| 1041 | if not os.path.exists(script_path): |
| 1042 | raise ValueError("Script file not found for this command") |
| 1043 | |
| 1044 | module_globals = runpy.run_path(script_path) |
| 1045 | hook = module_globals.get("run") |
| 1046 | if not callable(hook): |
| 1047 | raise ValueError('Script command must expose a callable "run(payload)" function') |
| 1048 | |
| 1049 | context = _get_context(context_id) |
| 1050 | history = _extract_chat_history(context) if command.get("include_history") else [] |
| 1051 | payload = { |
| 1052 | "command": _public_command_payload(command), |
| 1053 | "invocation": invocation, |
| 1054 | "arguments": invocation.get("arguments", {}), |
| 1055 | "context": { |
| 1056 | "context_id": context_id, |
| 1057 | "project_name": project_name, |
| 1058 | "agent": getattr(context, "agent0", None) if context else None, |
| 1059 | "chat_history": history, |
| 1060 | }, |
| 1061 | } |
| 1062 | |
| 1063 | result = hook(payload) |
| 1064 | if inspect.isawaitable(result): |
| 1065 | result = await result |
| 1066 | return _normalize_script_result(result) |
| 1067 | |
| 1068 | |
| 1069 | def _normalize_script_result(result: Any) -> dict[str, Any]: |
| 1070 | if isinstance(result, str): |
| 1071 | return {"text": result, "effects": []} |
| 1072 | |
| 1073 | if isinstance(result, dict): |
| 1074 | text = result.get("text") |
| 1075 | if text is None: |
| 1076 | text = result.get("replacement_text") |
| 1077 | |
| 1078 | effects = result.get("effects") |
| 1079 | if effects is None: |
| 1080 | effects = [] |
| 1081 | if not isinstance(effects, list): |
| 1082 | raise ValueError("Script result.effects must be an array when provided") |
| 1083 | |
| 1084 | normalized_text = str(text) if text is not None else "" |
| 1085 | return {"text": normalized_text, "effects": effects} |
| 1086 | |
| 1087 | raise ValueError("Script run(payload) must return either a string or an object") |
| 1088 | |
| 1089 | |
| 1090 | def _extract_chat_history(context: AgentContext | None) -> list[Any]: |
| 1091 | if not context: |
| 1092 | return [] |
| 1093 | |
| 1094 | for attribute in ("chat_history", "history", "messages"): |
| 1095 | value = getattr(context, attribute, None) |
| 1096 | if isinstance(value, list): |
| 1097 | return value |
| 1098 | |
| 1099 | getter = getattr(context, "get_data", None) |
| 1100 | if callable(getter): |
| 1101 | for key in ("chat_history", "messages", "history"): |
| 1102 | value = getter(key) |
| 1103 | if isinstance(value, list): |
| 1104 | return value |
| 1105 | |
| 1106 | return [] |
| 1107 | |
| 1108 | |
| 1109 | def _public_command_payload(command: dict[str, Any]) -> dict[str, Any]: |
| 1110 | payload = dict(command) |
| 1111 | payload.pop("body", None) |
| 1112 | return payload |
| 1113 | |
| 1114 | |
| 1115 | def _normalize_client_path(path: str) -> str: |
| 1116 | return files.normalize_a0_path(path).replace("\\", "/") |
| 1117 | |
| 1118 | |
| 1119 | def _paths_equal(path_a: str, path_b: str) -> bool: |
| 1120 | if not path_a or not path_b: |
| 1121 | return False |
| 1122 | return os.path.normcase(os.path.normpath(path_a)) == os.path.normcase( |
| 1123 | os.path.normpath(path_b) |
| 1124 | ) |
| 1125 | |
| 1126 | |
| 1127 | def _get_context(context_id: str = "") -> AgentContext | None: |
| 1128 | if context_id: |
| 1129 | return AgentContext.get(context_id) |
| 1130 | return AgentContext.current() or AgentContext.first() |
| 1131 | |
| 1132 | |
| 1133 | def _to_abs_path(path: str) -> str: |
| 1134 | return files.fix_dev_path(path) |
| 1135 | |
| 1136 | |
| 1137 | def _plugin_scope_label(plugin_name: str) -> str: |
| 1138 | return f"Plugin: {plugin_name}" |
| 1139 | |
| 1140 | |
| 1141 | def _mark_plugin_command(command: dict[str, Any], plugin_name: str) -> dict[str, Any]: |
| 1142 | scope_label = _plugin_scope_label(plugin_name) |
| 1143 | command["source_plugin"] = plugin_name |
| 1144 | command["scope_key"] = "plugin" |
| 1145 | command["scope_label"] = scope_label |
| 1146 | command["source_scope_key"] = "plugin" |
| 1147 | command["source_scope_label"] = scope_label |
| 1148 | return command |
| 1149 | |
| 1150 | |
| 1151 | def _mark_builtin_command(command: dict[str, Any]) -> dict[str, Any]: |
| 1152 | command["source_plugin"] = PLUGIN_NAME |
| 1153 | command["scope_key"] = "builtin" |
| 1154 | command["scope_label"] = "Built-in" |
| 1155 | command["source_scope_key"] = "builtin" |
| 1156 | command["source_scope_label"] = "Built-in" |
| 1157 | return command |
| 1158 | |
| 1159 | |
| 1160 | def _builtin_commands_dir() -> str: |
| 1161 | plugin_dir = plugins.find_plugin_dir(PLUGIN_NAME) |
| 1162 | return files.get_abs_path(plugin_dir, COMMANDS_DIR) if plugin_dir else "" |
| 1163 | |
| 1164 | |
| 1165 | def _is_builtin_command_path(path: str) -> bool: |
| 1166 | commands_dir = _builtin_commands_dir() |
| 1167 | return bool(commands_dir and files.is_in_dir(_to_abs_path(path), commands_dir)) |
| 1168 | |
| 1169 | |
| 1170 | def _plugin_name_for_commands_path(path: str) -> str: |
| 1171 | abs_path = _to_abs_path(path) |
| 1172 | for plugin_name in plugins.get_enabled_plugins(None): |
| 1173 | if plugin_name == PLUGIN_NAME: |
| 1174 | continue |
| 1175 | plugin_dir = plugins.find_plugin_dir(plugin_name) |
| 1176 | if not plugin_dir: |
| 1177 | continue |
| 1178 | plugin_commands_dir = files.get_abs_path(plugin_dir, COMMANDS_DIR) |
| 1179 | if files.is_in_dir(abs_path, plugin_commands_dir): |
| 1180 | return plugin_name |
| 1181 | return "" |
| 1182 | |
| 1183 | |
| 1184 | def _is_plugin_commands_dir(path: str) -> bool: |
| 1185 | """Check if a path is inside any installed plugin's commands/ subdirectory.""" |
| 1186 | return bool(_plugin_name_for_commands_path(path)) |
| 1187 | |
| 1188 | |
| 1189 | def strip_private_scope(scope: dict[str, Any]) -> dict[str, Any]: |
| 1190 | """Return a copy of *scope* with all keys prefixed by ``_`` removed.""" |
| 1191 | return {key: value for key, value in scope.items() if not key.startswith("_")} |
| 1192 | |
| 1193 | |
| 1194 | def _strip_private_scope(scope: dict[str, Any]) -> dict[str, Any]: |
| 1195 | return strip_private_scope(scope) |