Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 33 additions & 8 deletions src/agentpeek/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,24 +26,23 @@ class SettingsBundle:
status_line: Mapping[str, str] | None
skip_auto_permission_prompt: bool
# `policy-limits.json` from this scope, flattened to
# {restriction_name: allowed_bool}. Empty when the file is absent.
policy_restrictions: Mapping[str, bool]
# {restriction_name: "allowed" / "denied" / "denied — <message>"}.
# Empty when the file is absent.
policy_restrictions: Mapping[str, str]
# Enterprise-pushed remote-settings.json extras. Surfaced so users
# can see what their managed config injected into the session.
company_announcements: tuple[str, ...]
spinner_tips: tuple[str, ...]
# Relative paths inside `<root>/local/` — typically user-applied
# patches or scratch scripts that aren't Claude Code config but
# may shadow or modify the canonical hooks. Surfaced so the user
# can see what's there.
local_overrides: tuple[str, ...]
env: Mapping[str, str]
permissions_allow: tuple[str, ...]
permissions_deny: tuple[str, ...]
permissions_ask: tuple[str, ...]
enabled_plugins: tuple[str, ...]
hooks_raw: Mapping[str, tuple[Mapping[str, object], ...]]
hooks_dir_files: int
# Filenames (relative to `<root>/hooks/`) present on disk —
# surfaced as a list so users can see actual script names without
# opening the filesystem.
hooks_dir_files: tuple[str, ...]


@dataclass(frozen=True, slots=True)
Expand Down Expand Up @@ -96,6 +95,17 @@ class PluginManifest:
homepage: str | None
license: str | None
keywords: tuple[str, ...] = ()
# Declared name (`name` in plugin.json) — may differ from the
# install directory name. Surfaced so a renamed/forked plugin
# is recognizable.
name: str | None = None
# Upstream source repo URL. Accepts both bare-string form
# ("https://github.com/...") and the dict form ({"type": "git",
# "url": "..."}); normalized to a URL string here.
repository: str | None = None
# `requires` from plugin.json — qualified ids of plugins this
# plugin depends on. Empty when not declared.
requires: tuple[str, ...] = ()


@dataclass(frozen=True, slots=True)
Expand All @@ -105,6 +115,17 @@ class PluginSkill:
description: str | None
body: str
source_plugin: str | None = None
# `when_to_use` — trigger guidance shown to Claude Code so it
# knows when to auto-load this skill. Often more descriptive
# than `description`; surfaced here so users can see the
# activation contract.
when_to_use: str | None = None
# True/False if `user-invocable` is set in frontmatter; None
# when the field is absent (defaults vary).
user_invocable: bool | None = None
# Comma-separated tool allow/deny lists from frontmatter.
allowed_tools: tuple[str, ...] = ()
disallowed_tools: tuple[str, ...] = ()


@dataclass(frozen=True, slots=True)
Expand All @@ -114,6 +135,10 @@ class PluginAgent:
description: str | None
body: str
source_plugin: str | None = None
when_to_use: str | None = None
user_invocable: bool | None = None
allowed_tools: tuple[str, ...] = ()
disallowed_tools: tuple[str, ...] = ()


@dataclass(frozen=True, slots=True)
Expand Down
68 changes: 68 additions & 0 deletions src/agentpeek/parsers/plugin_contents.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
`source_plugin = qualified_id` for provenance.
"""

from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
Expand Down Expand Up @@ -114,6 +115,22 @@ def _parse_manifest(
if isinstance(keywords_raw, list)
else ()
)
requires_raw = d.get("requires")
requires: tuple[str, ...] = (
tuple(s for s in cast("list[object]", requires_raw) if isinstance(s, str))
if isinstance(requires_raw, list)
else ()
)
# `repository` accepts either a bare URL string or a dict like
# {"type": "git", "url": "..."}. Normalize to the URL.
repository_raw = d.get("repository")
repository: str | None = None
if isinstance(repository_raw, str):
repository = repository_raw
elif isinstance(repository_raw, dict):
repository = as_str(
cast("dict[str, object]", repository_raw).get("url")
)
manifest = PluginManifest(
description=as_str(d.get("description")),
version=as_str(d.get("version")),
Expand All @@ -122,6 +139,9 @@ def _parse_manifest(
homepage=as_str(d.get("homepage")),
license=as_str(d.get("license")),
keywords=keywords,
name=as_str(d.get("name")),
repository=repository,
requires=requires,
)
return d, manifest

Expand Down Expand Up @@ -160,6 +180,19 @@ def _parse_skills(
),
body=file.body,
source_plugin=qualified_id,
when_to_use=read_str_field(
file.metadata, "when_to_use",
path=skill_md, category="plugin_skill", warnings=warnings,
),
user_invocable=_as_optional_bool(file.metadata.get("user-invocable")),
allowed_tools=_split_csv_field(
file.metadata, "allowed-tools",
path=skill_md, category="plugin_skill", warnings=warnings,
),
disallowed_tools=_split_csv_field(
file.metadata, "disallowed-tools",
path=skill_md, category="plugin_skill", warnings=warnings,
),
)
)
return tuple(results)
Expand Down Expand Up @@ -191,11 +224,46 @@ def _parse_agents(
),
body=file.body,
source_plugin=qualified_id,
when_to_use=read_str_field(
file.metadata, "when_to_use",
path=md, category="plugin_agent", warnings=warnings,
),
user_invocable=_as_optional_bool(file.metadata.get("user-invocable")),
allowed_tools=_split_csv_field(
file.metadata, "allowed-tools",
path=md, category="plugin_agent", warnings=warnings,
),
disallowed_tools=_split_csv_field(
file.metadata, "disallowed-tools",
path=md, category="plugin_agent", warnings=warnings,
),
)
)
return tuple(results)


def _as_optional_bool(v: object) -> bool | None:
if isinstance(v, bool):
return v
return None


def _split_csv_field(
metadata: Mapping[str, object],
key: str,
*,
path: Path,
category: str,
warnings: list[ScanWarning],
) -> tuple[str, ...]:
raw = read_str_field(
metadata, key, path=path, category=category, warnings=warnings
)
if not raw:
return ()
return tuple(s.strip() for s in raw.split(",") if s.strip())


# --- Slash commands ------------------------------------------------------


Expand Down
130 changes: 82 additions & 48 deletions src/agentpeek/sources/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,21 @@ def _scan_settings(
as_str_tuple(user_perms.get("ask")),
as_str_tuple(local_perms.get("ask")),
)
# Same pattern in both allow and deny is a contradiction the
# user probably didn't mean — flag it. Claude Code's runtime
# resolution for this case isn't documented; we report what's
# in the files and let the user decide which one to remove.
for collision in sorted(set(permissions_allow) & set(permissions_deny)):
warnings.append(
ScanWarning(
path=user_path if user_path.exists() else local_path,
category="settings",
reason=(
f"permission rule {collision!r} appears in both "
"allow and deny"
),
)
)

user_plugins = as_dict(user_data.get("enabledPlugins")) or {}
local_plugins = as_dict(local_data.get("enabledPlugins")) or {}
Expand All @@ -123,19 +138,31 @@ def _scan_settings(
enabled_set.add(str(k))
enabled_plugins = tuple(sorted(enabled_set))

hooks_raw_dict = as_dict(user_data.get("hooks")) or {}
# Union hooks across user + local settings (matches the
# permissions / enabledPlugins precedent — Claude Code stacks
# both). Entries from local follow entries from user within
# each event so the original ordering is preserved.
hooks_raw: dict[str, tuple[dict[str, object], ...]] = {}
for event, entries in hooks_raw_dict.items():
if isinstance(entries, list):
event_entries: list[dict[str, object]] = []
merged_events: dict[str, list[dict[str, object]]] = {}
for source in (
as_dict(user_data.get("hooks")) or {},
as_dict(local_data.get("hooks")) or {},
):
for event, entries in source.items():
if not isinstance(entries, list):
continue
bucket = merged_events.setdefault(str(event), [])
for e in cast("list[object]", entries):
if isinstance(e, dict):
event_entries.append(cast("dict[str, object]", e))
hooks_raw[str(event)] = tuple(event_entries)
bucket.append(cast("dict[str, object]", e))
for event, bucket in merged_events.items():
hooks_raw[event] = tuple(bucket)

hooks_dir = root / "hooks"
hooks_dir_files = (
sum(1 for _ in hooks_dir.iterdir()) if hooks_dir.is_dir() else 0
hooks_dir_files: tuple[str, ...] = (
tuple(sorted(f.name for f in hooks_dir.iterdir() if f.is_file()))
if hooks_dir.is_dir()
else ()
)

# statusLine can be a dict ({"type": "command", "command": "..."})
Expand All @@ -144,19 +171,23 @@ def _scan_settings(
_parse_status_line(user_data.get("statusLine"))
)

skip_prompt = bool(
local_data.get("skipAutoPermissionPrompt")
or user_data.get("skipAutoPermissionPrompt")
)
# `in` rather than `or` — `or` short-circuits on False, so a
# project explicitly setting `skipAutoPermissionPrompt: false`
# would never override a user-level True. Use presence to
# distinguish "key absent" from "explicit False".
if "skipAutoPermissionPrompt" in local_data:
skip_prompt = bool(local_data["skipAutoPermissionPrompt"])
elif "skipAutoPermissionPrompt" in user_data:
skip_prompt = bool(user_data["skipAutoPermissionPrompt"])
else:
skip_prompt = False

policy_restrictions = _load_policy_limits(root, warnings)

company_announcements = as_str_tuple(remote_data.get("companyAnnouncements"))
spinner_override = as_dict(remote_data.get("spinnerTipsOverride")) or {}
spinner_tips = as_str_tuple(spinner_override.get("tips"))

local_overrides = _scan_local_overrides(root)

return SettingsBundle(
user_settings_path=user_path if user_path.exists() else None,
local_settings_path=local_path if local_path.exists() else None,
Expand All @@ -175,7 +206,6 @@ def _scan_settings(
policy_restrictions=MappingProxyType(policy_restrictions),
company_announcements=company_announcements,
spinner_tips=spinner_tips,
local_overrides=local_overrides,
hooks_raw=MappingProxyType(hooks_raw),
hooks_dir_files=hooks_dir_files,
)
Expand Down Expand Up @@ -679,31 +709,14 @@ def _flatten_marketplace(entry: dict[str, object]) -> dict[str, str]:
return flat


def _scan_local_overrides(root: Path) -> tuple[str, ...]:
"""Return relative paths of every file under `<root>/local/`.

Claude Code itself doesn't define a `local/` directory — it's a
user convention for staging patches or scratch scripts. We list
what's there so users see what's adjacent to their config; we
make no claim about what these files do.
"""
local_dir = root / "local"
if not local_dir.is_dir():
return ()
paths: list[str] = []
for f in sorted(local_dir.rglob("*")):
if f.is_file():
try:
paths.append(str(f.relative_to(local_dir)))
except ValueError:
continue
return tuple(paths)


def _load_policy_limits(root: Path, warnings: list[ScanWarning]) -> dict[str, bool]:
"""Return {restriction_name: allowed} from `<root>/policy-limits.json`.
def _load_policy_limits(root: Path, warnings: list[ScanWarning]) -> dict[str, str]:
"""Return {restriction_name: formatted-string} from
`<root>/policy-limits.json`.

Empty dict when the file is absent (typical for project scope).
The string includes both the allowed bool and any `message` /
`text` field the enterprise admin attached. Format:
`"allowed"` / `"denied"` / `"denied — <message>"` etc.
Empty dict when the file is absent.
"""
path = root / "policy-limits.json"
if not path.is_file():
Expand All @@ -716,28 +729,35 @@ def _load_policy_limits(root: Path, warnings: list[ScanWarning]) -> dict[str, bo
restrictions = cast("dict[str, object]", data).get("restrictions")
if not isinstance(restrictions, dict):
return {}
out: dict[str, bool] = {}
out: dict[str, str] = {}
for name, entry in cast("dict[str, object]", restrictions).items():
if isinstance(entry, dict):
allowed = cast("dict[str, object]", entry).get("allowed")
if isinstance(allowed, bool):
out[str(name)] = allowed
if not isinstance(entry, dict):
continue
entry_d = cast("dict[str, object]", entry)
allowed = entry_d.get("allowed")
verdict = (
"allowed" if allowed is True else "denied" if allowed is False else "?"
)
message = as_str(entry_d.get("message")) or as_str(entry_d.get("text"))
out[str(name)] = f"{verdict} — {message}" if message else verdict
return out


def _parse_status_line(v: object) -> Mapping[str, str] | None:
"""Normalize statusLine to a {type, command, ...} dict, or None.

Accepts dict-shaped configs (modern) and bare strings (legacy form
where the entire value is the shell command).
where the entire value is the shell command). Non-string dict
values (e.g. `padding: 1`, `refreshInterval: 30`) are coerced to
str rather than dropped — they're real config the user wrote and
inspecting them in agentpeek shouldn't silently lose them.
"""
if isinstance(v, str):
return MappingProxyType({"type": "command", "command": v})
if isinstance(v, dict):
d = {
str(k): str(val)
for k, val in cast("dict[str, object]", v).items()
if isinstance(val, str)
}
return MappingProxyType(d) if d else None
return None
Expand Down Expand Up @@ -769,8 +789,22 @@ def _load_blocklist(root: Path, warnings: list[ScanWarning]) -> dict[str, str]:
qid = as_str(entry_d.get("plugin"))
if not qid:
continue
reason = as_str(entry_d.get("reason")) or as_str(entry_d.get("text")) or ""
out[qid] = reason
reason = as_str(entry_d.get("reason"))
text = as_str(entry_d.get("text"))
if reason and text and reason != text:
warnings.append(
ScanWarning(
path=path,
category="plugins",
reason=(
f"blocklist entry for {qid} has both `reason` and "
f"`text` with different values; surfacing both"
),
)
)
out[qid] = f"{reason} (text: {text})"
else:
out[qid] = reason or text or ""
return out


Expand Down
Loading