provider_commands.py
Rama main · réplica de lectura del repositorio original.
"""Bounded, display-safe provider command inventories.
Provider discovery returns filesystem paths and other provider-owned metadata.
This module keeps that material on AgentsServer and projects only the small
allowlist needed by clients. A client selects an opaque id plus the inventory
revision; it never sends a path or provider command name back to the server.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import os
import re
import stat
import time
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
MAX_PROVIDER_COMMANDS = 512
MAX_PROVIDER_COMMAND_NAME_CHARS = 128
MAX_PROVIDER_COMMAND_LABEL_CHARS = 160
MAX_PROVIDER_COMMAND_DESCRIPTION_CHARS = 800
MAX_PROVIDER_COMMAND_PATH_CHARS = 4096
# OpenCode skills are discovered directly from its documented local defaults.
# These ceilings make opening the slash palette a bounded filesystem read even
# when a workspace contains a deliberately hostile directory tree.
MAX_OPENCODE_SKILL_ANCESTORS = 64
MAX_OPENCODE_SKILL_ROOTS = 3 * MAX_OPENCODE_SKILL_ANCESTORS + 3
MAX_OPENCODE_SKILL_SCAN_DEPTH = 8
MAX_OPENCODE_SKILL_SCAN_DIRS = 4096
MAX_OPENCODE_SKILL_SCAN_ENTRIES = 16384
MAX_OPENCODE_SKILL_FILE_BYTES = 1024 * 1024
MAX_OPENCODE_SKILL_TOTAL_BYTES = 16 * 1024 * 1024
MAX_OPENCODE_SKILL_FRONTMATTER_LINES = 256
OPENCODE_SKILL_SCAN_TIMEOUT_SECONDS = 2.0
_COMMAND_NAME_RE = re.compile(
rf"[A-Za-z0-9_][A-Za-z0-9_.:-]{{0,{MAX_PROVIDER_COMMAND_NAME_CHARS - 1}}}"
)
_OPENCODE_SKILL_NAME_RE = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*")
_FRONTMATTER_KEY_RE = re.compile(r"([A-Za-z][A-Za-z0-9_-]*):(?:[ \t]*(.*))?")
_LOCATION_RE = re.compile(
# Provider metadata is optional display copy, so redact the entire suffix
# after a URL/path marker. Local paths may contain spaces and punctuation;
# trying to guess their endpoint can expose the unredacted tail.
r"(?P<url>[A-Za-z][A-Za-z0-9+.-]*://[\s\S]*)"
r"|(?P<path>(?<![A-Za-z0-9_])(?:~[/\\]|\.\.?[/\\]|[A-Za-z]:[/\\]|/|\\\\)[\s\S]*)"
)
_EMAIL_RE = re.compile(
r"(?<![A-Za-z0-9.!#$%&'*+/=?^_`{|}~-])"
r"[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+@"
r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,251}[A-Za-z0-9])?"
r"(?![A-Za-z0-9.-])"
)
_UNSAFE_CATEGORIES = frozenset({"Cc", "Cf", "Cs", "Co", "Cn", "Zl", "Zp"})
_BIDI_CONTROLS = frozenset({
"\u061c",
"\u200e",
"\u200f",
"\u202a",
"\u202b",
"\u202c",
"\u202d",
"\u202e",
"\u2066",
"\u2067",
"\u2068",
"\u2069",
})
@dataclass(frozen=True)
class ProviderCommandRecord:
public: dict[str, Any]
native: dict[str, str]
@dataclass(frozen=True)
class ProviderCommandInventory:
backend: str
revision: str
records: tuple[ProviderCommandRecord, ...]
truncated: bool = False
@property
def commands(self) -> list[dict[str, Any]]:
return [dict(record.public) for record in self.records]
def resolve(self, command_id: str) -> ProviderCommandRecord | None:
for record in self.records:
if record.public["id"] == command_id:
return record
return None
class ProviderCommandDiscoveryError(RuntimeError):
"""A bounded local provider inventory could not be read safely."""
@dataclass
class _OpenCodeSkillScanBudget:
deadline: float
directories: int = 0
entries: int = 0
files: int = 0
total_bytes: int = 0
def check(self) -> None:
if time.monotonic() > self.deadline:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its time limit"
)
def count_directory(self) -> None:
self.check()
self.directories += 1
if self.directories > MAX_OPENCODE_SKILL_SCAN_DIRS:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its directory limit"
)
def count_file(self, size: int) -> None:
self.check()
self.files += 1
if self.files > MAX_PROVIDER_COMMANDS:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its file limit"
)
if size < 0 or size > MAX_OPENCODE_SKILL_FILE_BYTES:
raise ProviderCommandDiscoveryError(
"an OpenCode skill exceeds the per-file byte limit"
)
self.total_bytes += size
if self.total_bytes > MAX_OPENCODE_SKILL_TOTAL_BYTES:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its aggregate byte limit"
)
def count_entry(self) -> None:
self.check()
self.entries += 1
if self.entries > MAX_OPENCODE_SKILL_SCAN_ENTRIES:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its entry limit"
)
def canonical_provider_command_name(value: Any) -> str | None:
if not isinstance(value, str):
return None
candidate = unicodedata.normalize("NFC", value)
if candidate != value or candidate != candidate.strip():
return None
if _COMMAND_NAME_RE.fullmatch(candidate) is None:
return None
return candidate
def _replace_local_path(match: re.Match[str]) -> str:
url = match.group("url")
if url is not None:
# Even HTTP(S) user info, hosts, paths and query parameters can contain
# credentials or personally identifying local infrastructure data.
return "<url>"
return "<path>"
def sanitize_provider_command_text(value: Any, max_chars: int) -> str:
if not isinstance(value, str):
return ""
normalized = unicodedata.normalize("NFC", value)
safe = "".join(
character
for character in normalized
if character not in _BIDI_CONTROLS
and (
character in {"\n", "\r", "\t"}
or unicodedata.category(character) not in _UNSAFE_CATEGORIES
)
)
safe = _EMAIL_RE.sub("<email>", safe)
safe = _LOCATION_RE.sub(_replace_local_path, safe)
safe = " ".join(safe.split())
if len(safe) <= max_chars:
return safe
return safe[: max(0, max_chars - 1)].rstrip() + "…"
def _plain_frontmatter_scalar(value: str) -> str | None:
"""Parse the conservative scalar subset used by the local skill index.
OpenCode uses a complete YAML parser. AgentsServer intentionally accepts a
smaller, deterministic subset for the two display fields it needs rather
than adding executable tags, aliases, or another deployment dependency.
Files using richer YAML remain usable in OpenCode itself but are omitted
from this bounded palette.
"""
candidate = value.strip()
if not candidate:
return ""
if candidate.startswith('"'):
try:
decoded = json.loads(candidate)
except (TypeError, json.JSONDecodeError):
return None
return decoded if isinstance(decoded, str) else None
if candidate.startswith("'"):
if len(candidate) < 2 or not candidate.endswith("'"):
return None
return candidate[1:-1].replace("''", "'")
if candidate[0] in "[{&*!|>":
return None
# A hash starts a YAML comment only when separated from the scalar.
candidate = re.split(r"[ \t]+#", candidate, maxsplit=1)[0].rstrip()
if re.search(r":[ \t]", candidate):
return None
if candidate.lower() in {"null", "true", "false", "~"}:
return None
return candidate
def _opencode_skill_frontmatter(data: bytes) -> tuple[str, str, str] | None:
try:
text = data.decode("utf-8", errors="strict")
except UnicodeDecodeError:
return None
if text.startswith("\ufeff"):
return None
lines = text.splitlines()
if not lines or lines[0].strip() != "---":
return None
closing = None
for index, line in enumerate(
lines[1 : MAX_OPENCODE_SKILL_FRONTMATTER_LINES + 1],
start=1,
):
if line.strip() == "---":
closing = index
break
if closing is None:
return None
values: dict[str, str] = {}
body_lines = lines[1:closing]
index = 0
while index < len(body_lines):
line = body_lines[index]
index += 1
if not line.strip() or line.lstrip().startswith("#"):
continue
# Nested metadata and other unknown structured fields are irrelevant.
# A recognized field must be a top-level, single-line scalar.
if line[:1].isspace():
continue
match = _FRONTMATTER_KEY_RE.fullmatch(line)
if match is None:
continue
key, raw_value = match.groups()
if key not in {"name", "description"}:
continue
if key in values or raw_value is None:
return None
if key == "description" and raw_value.strip() in {
"|", "|-", "|+", ">", ">-", ">+",
}:
block: list[str] = []
while index < len(body_lines):
block_line = body_lines[index]
if block_line and not block_line[:1].isspace():
break
index += 1
block.append(block_line)
nonblank = [value for value in block if value.strip()]
if not nonblank or any(value.startswith("\t") for value in nonblank):
return None
indentation = min(len(value) - len(value.lstrip(" ")) for value in nonblank)
if indentation <= 0:
return None
value = "\n".join(
item[indentation:] if item.strip() else ""
for item in block
).strip("\n")
values[key] = value
continue
value = _plain_frontmatter_scalar(raw_value)
if value is None:
return None
values[key] = value
name = values.get("name", "")
description = values.get("description", "")
if (
not 1 <= len(name) <= 64
or _OPENCODE_SKILL_NAME_RE.fullmatch(name) is None
or not 1 <= len(description) <= 1024
):
return None
# OpenCode's native Skill loader passes gray-matter's trimmed Markdown
# content to the model, not the YAML frontmatter itself.
content = "\n".join(lines[closing + 1:]).strip()
return name, description, content
def _path_has_symlink(path: Path) -> bool:
"""Return true when any existing component is a symbolic link."""
absolute = Path(os.path.abspath(os.path.normpath(str(path))))
parts = absolute.parts
if not parts:
return True
current = Path(parts[0])
for part in parts[1:]:
current /= part
try:
if stat.S_ISLNK(current.lstat().st_mode):
return True
except OSError:
return True
return False
def _read_regular_file_without_symlinks(
path: Path,
budget: _OpenCodeSkillScanBudget,
) -> bytes:
if _path_has_symlink(path):
raise ProviderCommandDiscoveryError(
"an OpenCode skill path contains a symbolic link"
)
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
try:
descriptor = os.open(path, flags)
except OSError as exc:
raise ProviderCommandDiscoveryError(
"an OpenCode skill could not be opened safely"
) from exc
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode):
raise ProviderCommandDiscoveryError(
"an OpenCode skill is not a regular file"
)
budget.count_file(before.st_size)
chunks: list[bytes] = []
remaining = before.st_size
while remaining:
budget.check()
chunk = os.read(descriptor, min(64 * 1024, remaining))
if not chunk:
raise ProviderCommandDiscoveryError(
"an OpenCode skill changed while being read"
)
chunks.append(chunk)
remaining -= len(chunk)
data = b"".join(chunks)
after = os.fstat(descriptor)
if (
before.st_dev,
before.st_ino,
before.st_size,
before.st_mtime_ns,
) != (
after.st_dev,
after.st_ino,
after.st_size,
after.st_mtime_ns,
):
raise ProviderCommandDiscoveryError(
"an OpenCode skill changed while being read"
)
return data
finally:
os.close(descriptor)
def _bounded_skill_files(
root: Path,
budget: _OpenCodeSkillScanBudget,
) -> list[Path]:
if _path_has_symlink(root):
return []
try:
root_stat = root.lstat()
except FileNotFoundError:
return []
except OSError as exc:
raise ProviderCommandDiscoveryError(
"an OpenCode skill root could not be inspected"
) from exc
if not stat.S_ISDIR(root_stat.st_mode):
return []
matches: list[Path] = []
pending: list[tuple[Path, int]] = [(root, 0)]
while pending:
directory, depth = pending.pop()
budget.count_directory()
try:
with os.scandir(directory) as iterator:
entries = []
for entry in iterator:
budget.count_entry()
entries.append(entry)
entries.sort(key=lambda item: item.name)
except OSError as exc:
raise ProviderCommandDiscoveryError(
"an OpenCode skill directory could not be read"
) from exc
child_dirs: list[Path] = []
for entry in entries:
budget.check()
try:
entry_stat = entry.stat(follow_symlinks=False)
except OSError as exc:
raise ProviderCommandDiscoveryError(
"an OpenCode skill entry could not be inspected"
) from exc
if stat.S_ISLNK(entry_stat.st_mode):
continue
candidate = directory / entry.name
if stat.S_ISDIR(entry_stat.st_mode):
if depth < MAX_OPENCODE_SKILL_SCAN_DEPTH:
child_dirs.append(candidate)
continue
if stat.S_ISREG(entry_stat.st_mode) and entry.name == "SKILL.md":
matches.append(candidate)
# Stack reversal preserves lexical traversal while remaining iterative.
pending.extend((child, depth + 1) for child in reversed(child_dirs))
return matches
def _opencode_project_ancestors(cwd: Path) -> list[Path]:
"""Walk nearest-first through the git worktree, or only cwd outside git."""
ancestors: list[Path] = []
current = cwd
worktree_index: int | None = None
for _ in range(MAX_OPENCODE_SKILL_ANCESTORS):
ancestors.append(current)
git_marker = current / ".git"
try:
marker_mode = git_marker.lstat().st_mode
except FileNotFoundError:
marker_mode = 0
except OSError as exc:
raise ProviderCommandDiscoveryError(
"the OpenCode worktree boundary could not be inspected"
) from exc
if marker_mode and not stat.S_ISLNK(marker_mode):
worktree_index = len(ancestors) - 1
break
parent = current.parent
if parent == current:
break
current = parent
else:
raise ProviderCommandDiscoveryError(
"the OpenCode worktree exceeds the ancestor limit"
)
return ancestors[: worktree_index + 1] if worktree_index is not None else [cwd]
def opencode_provider_skill_inventory(
*,
cwd: str,
home: str,
xdg_config_home: str | None,
selector_secret: str,
binding_context: str,
limit: int = MAX_PROVIDER_COMMANDS,
timeout_seconds: float = OPENCODE_SKILL_SCAN_TIMEOUT_SECONDS,
) -> ProviderCommandInventory:
"""Index the documented default local OpenCode skill roots only.
This deliberately does not evaluate OpenCode configuration or start its
CLI. Configured paths/URLs, plugins, commands, agents, MCP prompts, the
undocumented ``~/.opencode`` compatibility root, and built-in skills are
outside this side-effect-free v1 contract.
"""
try:
canonical_cwd = Path(
os.path.abspath(os.path.normpath(os.path.expanduser(cwd)))
).resolve(strict=True)
canonical_home = Path(
os.path.abspath(os.path.normpath(os.path.expanduser(home)))
).resolve(strict=True)
except (OSError, RuntimeError, ValueError) as exc:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery boundaries are unavailable"
) from exc
if not canonical_cwd.is_dir() or not canonical_home.is_dir():
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery boundaries are not directories"
)
if xdg_config_home:
expanded_xdg = Path(os.path.expanduser(xdg_config_home))
config_home = (
expanded_xdg
if expanded_xdg.is_absolute()
else canonical_home / ".config"
)
else:
config_home = canonical_home / ".config"
try:
if config_home.exists():
config_home = config_home.resolve(strict=True)
except (OSError, RuntimeError) as exc:
raise ProviderCommandDiscoveryError(
"the OpenCode configuration boundary could not be resolved"
) from exc
project_roots: list[tuple[Path, str]] = []
for ancestor in _opencode_project_ancestors(canonical_cwd):
project_roots.extend((
(ancestor / ".opencode" / "skills", "project"),
(ancestor / ".claude" / "skills", "project"),
(ancestor / ".agents" / "skills", "project"),
))
roots = [
*project_roots,
(config_home / "opencode" / "skills", "user"),
(canonical_home / ".claude" / "skills", "user"),
(canonical_home / ".agents" / "skills", "user"),
]
if len(roots) > MAX_OPENCODE_SKILL_ROOTS:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its root limit"
)
budget = _OpenCodeSkillScanBudget(
deadline=time.monotonic() + max(0.01, float(timeout_seconds))
)
discovered: dict[str, list[tuple[Path, bytes, str, str]]] = {}
seen_paths: set[str] = set()
for root, scope in roots:
budget.check()
for path in _bounded_skill_files(root, budget):
canonical_path = os.path.abspath(os.path.normpath(str(path)))
if len(canonical_path) > MAX_PROVIDER_COMMAND_PATH_CHARS:
raise ProviderCommandDiscoveryError(
"an OpenCode skill path exceeds the path limit"
)
if canonical_path in seen_paths:
continue
seen_paths.add(canonical_path)
data = _read_regular_file_without_symlinks(path, budget)
parsed = _opencode_skill_frontmatter(data)
if parsed is None:
continue
name, description, _content = parsed
if path.parent.name != name:
continue
discovered.setdefault(name, []).append((path, data, scope, description))
bounded_limit = max(1, min(int(limit), MAX_PROVIDER_COMMANDS))
unique = [
(name, entries[0])
for name, entries in sorted(discovered.items())
if len(entries) == 1
]
if len(unique) > bounded_limit:
raise ProviderCommandDiscoveryError(
"OpenCode skill discovery exceeded its inventory limit"
)
records: list[ProviderCommandRecord] = []
for name, (path, data, scope, description) in unique:
canonical_path = os.path.abspath(os.path.normpath(str(path)))
digest = hashlib.sha256(data).hexdigest()
native_identity = json.dumps(
[canonical_path, digest],
ensure_ascii=False,
separators=(",", ":"),
)
command_id = _opaque_command_id(
selector_secret,
binding_context,
"opencode",
str(canonical_cwd),
name,
native_identity,
)
safe_description = sanitize_provider_command_text(
description,
MAX_PROVIDER_COMMAND_DESCRIPTION_CHARS,
)
records.append(ProviderCommandRecord(
public={
"id": command_id,
"name": name,
"label": name,
"description": safe_description,
"scope": scope,
"source": "opencode",
"kind": "skill",
"invocation": f"/{name}",
},
native={
"name": name,
"path": canonical_path,
"directory": os.path.dirname(canonical_path),
"content_sha256": digest,
},
))
return _inventory(
selector_secret,
binding_context,
"opencode",
str(canonical_cwd),
records,
truncated=False,
)
def validate_opencode_provider_skill_record(
record: ProviderCommandRecord,
) -> tuple[str, str, str]:
"""Revalidate one private skill identity immediately before launch."""
name = str(record.native.get("name") or "")
raw_path = str(record.native.get("path") or "")
expected_directory = str(record.native.get("directory") or "")
expected_digest = str(record.native.get("content_sha256") or "")
if (
record.public.get("source") != "opencode"
or record.public.get("kind") != "skill"
or record.public.get("name") != name
or record.public.get("invocation") != f"/{name}"
or _OPENCODE_SKILL_NAME_RE.fullmatch(name) is None
or not raw_path
or not os.path.isabs(raw_path)
or len(raw_path) > MAX_PROVIDER_COMMAND_PATH_CHARS
or not expected_directory
or not hmac.compare_digest(
expected_digest,
expected_digest.lower(),
)
or re.fullmatch(r"[0-9a-f]{64}", expected_digest) is None
):
raise ProviderCommandDiscoveryError(
"the selected OpenCode skill identity is invalid"
)
path = Path(os.path.abspath(os.path.normpath(raw_path)))
if _path_has_symlink(path):
raise ProviderCommandDiscoveryError(
"the selected OpenCode skill path is no longer safe"
)
budget = _OpenCodeSkillScanBudget(
deadline=time.monotonic() + OPENCODE_SKILL_SCAN_TIMEOUT_SECONDS
)
data = _read_regular_file_without_symlinks(path, budget)
parsed = _opencode_skill_frontmatter(data)
canonical_directory = os.path.dirname(str(path))
if (
parsed is None
or parsed[0] != name
or path.parent.name != name
or canonical_directory != expected_directory
or not hmac.compare_digest(
hashlib.sha256(data).hexdigest(),
expected_digest,
)
):
raise ProviderCommandDiscoveryError(
"the selected OpenCode skill changed before launch"
)
return name, canonical_directory, parsed[2]
def _opaque_command_id(
selector_secret: str,
binding_context: str,
backend: str,
cwd: str,
name: str,
native_identity: str,
) -> str:
payload = json.dumps(
[
"provider-command-v1",
binding_context,
backend,
cwd,
name,
native_identity,
],
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
return "pcmd_" + hmac.new(
selector_secret.encode("utf-8"),
payload,
hashlib.sha256,
).hexdigest()[:32]
def _inventory_revision(
selector_secret: str,
binding_context: str,
backend: str,
cwd: str,
records: Iterable[ProviderCommandRecord],
*,
truncated: bool,
) -> str:
payload = {
"version": 1,
"binding_context": binding_context,
"backend": backend,
"cwd": cwd,
"truncated": truncated,
# Provider order is presentation-only and may change across otherwise
# identical cold connects. A semantic inventory revision must not
# invalidate a queued selection solely because that order changed.
"commands": sorted(
(record.public for record in records),
key=lambda item: str(item.get("id") or ""),
),
}
encoded = json.dumps(
payload,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return "pcmdrev_" + hmac.new(
selector_secret.encode("utf-8"),
encoded,
hashlib.sha256,
).hexdigest()[:32]
def _inventory(
selector_secret: str,
binding_context: str,
backend: str,
cwd: str,
records: list[ProviderCommandRecord],
*,
truncated: bool,
) -> ProviderCommandInventory:
frozen = tuple(records)
return ProviderCommandInventory(
backend=backend,
revision=_inventory_revision(
selector_secret,
binding_context,
backend,
cwd,
frozen,
truncated=truncated,
),
records=frozen,
truncated=truncated,
)
def codex_provider_command_inventory(
value: Any,
*,
cwd: str,
selector_secret: str,
binding_context: str,
limit: int = MAX_PROVIDER_COMMANDS,
) -> ProviderCommandInventory:
"""Project one cwd entry from Codex ``skills/list``.
The provider-resolved SKILL.md path remains only in ``record.native`` and
is never part of ``record.public``.
"""
raw_entries = value.get("data") if isinstance(value, dict) else None
matching: dict[str, Any] | None = None
if isinstance(raw_entries, list):
normalized_cwd = str(Path(cwd).resolve())
for entry in raw_entries[:32]:
if not isinstance(entry, dict):
continue
raw_cwd = entry.get("cwd")
if not isinstance(raw_cwd, str):
continue
try:
candidate_cwd = str(Path(raw_cwd).resolve())
except (OSError, RuntimeError):
continue
if candidate_cwd == normalized_cwd:
matching = entry
break
raw_skills = matching.get("skills") if isinstance(matching, dict) else None
if not isinstance(raw_skills, list):
raw_skills = []
bounded_limit = max(1, min(int(limit), MAX_PROVIDER_COMMANDS))
records: list[ProviderCommandRecord] = []
seen_ids: set[str] = set()
scanned = 0
for raw in raw_skills:
if scanned >= bounded_limit:
break
scanned += 1
if not isinstance(raw, dict) or raw.get("enabled") is False:
continue
name = canonical_provider_command_name(raw.get("name"))
raw_path = raw.get("path")
if (
name is None
or not isinstance(raw_path, str)
or not raw_path
or len(raw_path) > MAX_PROVIDER_COMMAND_PATH_CHARS
or "\x00" in raw_path
or not Path(raw_path).is_absolute()
):
continue
interface = raw.get("interface")
if not isinstance(interface, dict):
interface = {}
label = sanitize_provider_command_text(
interface.get("displayName") or name,
MAX_PROVIDER_COMMAND_LABEL_CHARS,
) or name
description = sanitize_provider_command_text(
interface.get("shortDescription") or raw.get("description"),
MAX_PROVIDER_COMMAND_DESCRIPTION_CHARS,
)
raw_scope = str(raw.get("scope") or "").strip().lower()
scope = raw_scope if raw_scope in {"project", "user", "admin", "system"} else None
command_id = _opaque_command_id(
selector_secret,
binding_context,
"codex",
cwd,
name,
raw_path,
)
if command_id in seen_ids:
continue
seen_ids.add(command_id)
public = {
"id": command_id,
"name": name,
"label": label,
"description": description,
"scope": scope,
"source": "plugin" if raw.get("pluginId") else "codex",
"kind": "skill",
# AgentsDock presents one slash-command palette for every provider.
# The server converts this slash token into Codex's native text +
# structured skill input only after validating the opaque id.
"invocation": f"/{name}",
}
records.append(
ProviderCommandRecord(
public=public,
native={"name": name, "path": raw_path},
)
)
return _inventory(
selector_secret,
binding_context,
"codex",
cwd,
records,
truncated=len(raw_skills) > bounded_limit,
)
def claude_provider_command_inventory(
value: Any,
*,
cwd: str,
selector_secret: str,
binding_context: str,
control_generation: str | None = None,
limit: int = MAX_PROVIDER_COMMANDS,
) -> ProviderCommandInventory:
"""Project only Claude's provider-reported command metadata.
``get_server_info`` also contains account data. Callers should already
project it inside the SDK actor, and this second allowlist prevents an
accidental raw top-level return from reaching the API.
"""
raw_commands = value.get("commands") if isinstance(value, dict) else None
if not isinstance(raw_commands, list):
raw_commands = []
bounded_limit = max(1, min(int(limit), MAX_PROVIDER_COMMANDS))
records: list[ProviderCommandRecord] = []
seen_names: set[str] = set()
scanned = 0
for raw in raw_commands:
if scanned >= bounded_limit:
break
scanned += 1
if not isinstance(raw, dict):
continue
name = canonical_provider_command_name(raw.get("name"))
if name is None or name in seen_names:
continue
seen_names.add(name)
description = sanitize_provider_command_text(
raw.get("description"),
MAX_PROVIDER_COMMAND_DESCRIPTION_CHARS,
)
argument_hint = sanitize_provider_command_text(
raw.get("argumentHint"),
MAX_PROVIDER_COMMAND_LABEL_CHARS,
)
if argument_hint:
description = (
f"{description} {argument_hint}".strip()
)[:MAX_PROVIDER_COMMAND_DESCRIPTION_CHARS]
command_id = _opaque_command_id(
selector_secret,
binding_context,
"claude",
cwd,
name,
name,
)
public = {
"id": command_id,
"name": name,
"label": name,
"description": description,
"scope": None,
"source": "claude",
"kind": "command",
"invocation": f"/{name}",
}
records.append(
ProviderCommandRecord(
public=public,
native={
"name": name,
**(
{"control_generation": control_generation}
if control_generation
else {}
),
},
)
)
return _inventory(
selector_secret,
binding_context,
"claude",
cwd,
records,
truncated=bool(
isinstance(value, dict)
and value.get("_agentsdock_provider_commands_truncated")
)
or len(raw_commands) > bounded_limit,
)
def empty_provider_command_inventory(
backend: str,
*,
cwd: str,
selector_secret: str,
binding_context: str,
) -> ProviderCommandInventory:
return _inventory(
selector_secret,
binding_context,
backend,
cwd,
[],
truncated=False,
)