mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-07 17:33:33 -10:00
fix: make project-instruction hierarchy and rules command native to standalone praisonai-code (fixes #4337)
On a standalone `pip install praisonai-code` the subtree instruction hook, `--instructions` resolution, and the `rules` command silently degraded because their implementations were reachable only via the optional `praisonai_bot` / `praisonai` wrapper packages. - Add a resident, dependency-free `praisonai_code/integration/context_files.py` (stdlib + `praisonaiagents.hooks` only) so subtree AGENTS.md loading and `--instructions` glob/URL resolution work without the optional packages. - Wire `run.py::_wire_subtree_context_hook` to fall back to the resident helper after the optional bot/wrapper copies, removing the silent no-op. - Give `rules list/add/clear` a native implementation backed by core `praisonaiagents.memory.RulesManager`, delegating to the wrapper bridge only when core is genuinely unavailable. - Add standalone regression tests that block the optional packages. Co-authored-by: MervinPraison <[email protected]>
This commit is contained in:
1 parent
689978100b
commit
448b7bb912
5 files changed
+909
-28
No files matched your search
@@ -1,41 +1,130 @@
|
||||
"""
|
||||
Rules command group for PraisonAI CLI.
|
||||
|
||||
Provides rules management commands.
|
||||
Provides native rules management (list/add/clear) for the standalone
|
||||
``praisonai-code`` terminal product. The underlying storage is core's
|
||||
``praisonaiagents.memory.RulesManager`` (PRAISON.md / CLAUDE.md / AGENTS.md and
|
||||
``.praison/rules/*.md``), so these commands work without the optional
|
||||
``praisonai`` wrapper. The wrapper bridge is retained only as a fallback for
|
||||
environments where core is unexpectedly unavailable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
import typer
|
||||
|
||||
app = typer.Typer(help="Rules management")
|
||||
|
||||
|
||||
def _rules_manager():
|
||||
"""Return a core ``RulesManager`` for the current workspace, or ``None``.
|
||||
|
||||
Core (``praisonaiagents``) is a hard dependency of ``praisonai-code``, so
|
||||
this normally succeeds on a standalone install. ``None`` is returned only
|
||||
when core is genuinely unavailable, letting the caller fall back to the
|
||||
wrapper bridge with a single clear notice rather than crashing.
|
||||
"""
|
||||
try:
|
||||
from praisonaiagents.memory import RulesManager
|
||||
except ImportError:
|
||||
return None
|
||||
return RulesManager(workspace_path=os.getcwd())
|
||||
|
||||
|
||||
def _fallback(argv: list[str]) -> None:
|
||||
"""Delegate to the wrapper bridge for a rules subcommand (last resort)."""
|
||||
from praisonai_code._wrapper_bridge import run_wrapper_command
|
||||
|
||||
run_wrapper_command(argv, feature="rules")
|
||||
|
||||
|
||||
@app.command("list")
|
||||
def rules_list():
|
||||
"""List active rules."""
|
||||
from praisonai_code._wrapper_bridge import run_wrapper_command
|
||||
|
||||
argv = ['rules', 'list']
|
||||
|
||||
run_wrapper_command(argv, feature="rules")
|
||||
manager = _rules_manager()
|
||||
if manager is None:
|
||||
_fallback(['rules', 'list'])
|
||||
return
|
||||
|
||||
try:
|
||||
all_rules = manager.get_all_rules()
|
||||
except Exception as exc: # noqa: BLE001 - never crash the CLI on rules read
|
||||
typer.echo(f"Failed to load rules: {exc}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
if not all_rules:
|
||||
typer.echo(
|
||||
"No rules found. Create PRAISON.md, CLAUDE.md, AGENTS.md, "
|
||||
"or files in .praison/rules/"
|
||||
)
|
||||
return
|
||||
|
||||
for rule in all_rules:
|
||||
desc = rule.description or ""
|
||||
if len(desc) > 60:
|
||||
desc = desc[:57] + "..."
|
||||
suffix = f" - {desc}" if desc else ""
|
||||
typer.echo(f"{rule.name} [{rule.activation}]{suffix}")
|
||||
|
||||
|
||||
@app.command("add")
|
||||
def rules_add(
|
||||
rule: str = typer.Argument(..., help="Rule to add"),
|
||||
rule: str = typer.Argument(..., help="Rule name"),
|
||||
content: str = typer.Argument(None, help="Rule content"),
|
||||
):
|
||||
"""Add a rule."""
|
||||
from praisonai_code._wrapper_bridge import run_wrapper_command
|
||||
|
||||
argv = ['rules', 'add', rule]
|
||||
|
||||
run_wrapper_command(argv, feature="rules")
|
||||
"""Add a rule (workspace scope)."""
|
||||
manager = _rules_manager()
|
||||
if manager is None:
|
||||
argv = ['rules', 'add', rule]
|
||||
if content:
|
||||
argv.append(content)
|
||||
_fallback(argv)
|
||||
return
|
||||
|
||||
if not content:
|
||||
typer.echo(
|
||||
"Rule content required. Usage: praisonai rules add <name> <content>",
|
||||
err=True,
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
try:
|
||||
manager.create_rule(
|
||||
name=rule,
|
||||
content=content,
|
||||
description=f"Rule created via CLI: {rule}",
|
||||
activation="always",
|
||||
scope="workspace",
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 - surface a clean error, no traceback
|
||||
typer.echo(f"Failed to add rule '{rule}': {exc}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
typer.echo(f"Rule added: {rule}")
|
||||
|
||||
|
||||
@app.command("clear")
|
||||
def rules_clear():
|
||||
"""Clear all rules."""
|
||||
from praisonai_code._wrapper_bridge import run_wrapper_command
|
||||
|
||||
argv = ['rules', 'clear']
|
||||
|
||||
run_wrapper_command(argv, feature="rules")
|
||||
"""Clear all workspace rules created via the CLI."""
|
||||
manager = _rules_manager()
|
||||
if manager is None:
|
||||
_fallback(['rules', 'clear'])
|
||||
return
|
||||
|
||||
try:
|
||||
all_rules = manager.get_all_rules()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
typer.echo(f"Failed to load rules: {exc}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
removed = 0
|
||||
for rule in all_rules:
|
||||
try:
|
||||
if manager.delete_rule(rule.name):
|
||||
removed += 1
|
||||
except Exception: # noqa: BLE001 - skip undeletable (e.g. discovered) rules
|
||||
continue
|
||||
|
||||
typer.echo(f"Cleared {removed} rule(s).")
|
||||
@@ -2284,8 +2284,10 @@ def _wire_subtree_context_hook(
|
||||
``already_loaded`` so nested files are not duplicated.
|
||||
|
||||
No-op (leaving the default run unchanged) when the up-front rules are
|
||||
disabled via ``--no-rules`` / ``PRAISON_NO_RULES`` or when the optional
|
||||
helper is unavailable. Honours the existing ``PRAISON_CONTEXT_BUDGET``
|
||||
disabled via ``--no-rules`` / ``PRAISON_NO_RULES``. The subtree-context
|
||||
helper is now resident in ``praisonai_code`` so it works on a standalone
|
||||
``pip install praisonai-code`` without the optional ``praisonai_bot`` /
|
||||
``praisonai`` packages. Honours the existing ``PRAISON_CONTEXT_BUDGET``
|
||||
character budget shared with the interactive path.
|
||||
"""
|
||||
import os
|
||||
@@ -2300,15 +2302,18 @@ def _wire_subtree_context_hook(
|
||||
try:
|
||||
import importlib
|
||||
|
||||
# The subtree-context helper is optional. Its canonical home is the
|
||||
# ``praisonai_bot`` support package; the ``praisonai`` wrapper only
|
||||
# re-exports it via a compat shim. Prefer ``praisonai_bot`` so the hook
|
||||
# is available on any install that ships it (matching ``chat``) without
|
||||
# forcing the full wrapper, then fall back to the wrapper shim. Either
|
||||
# way ``run.py`` avoids a module-level ``praisonai`` import (C7 gate).
|
||||
# The subtree-context helper is now resident in ``praisonai_code`` so
|
||||
# the standalone terminal product keeps first-class subtree instruction
|
||||
# loading and ``--instructions`` resolution without the optional
|
||||
# ``praisonai_bot`` support package or the ``praisonai`` wrapper (C7
|
||||
# standalone contract). For backward compatibility we still prefer an
|
||||
# installed ``praisonai_bot``/``praisonai`` copy when present (so a host
|
||||
# that customised it keeps its behaviour), falling back to the resident
|
||||
# implementation otherwise.
|
||||
context_files = None
|
||||
for _mod in ("praisonai_bot.integration.context_files",
|
||||
"praisonai.integration.context_files"):
|
||||
"praisonai.integration.context_files",
|
||||
"praisonai_code.integration.context_files"):
|
||||
try:
|
||||
context_files = importlib.import_module(_mod)
|
||||
break
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Resident integration helpers for the standalone praisonai-code CLI."""
|
||||
@@ -0,0 +1,603 @@
|
||||
"""AGENTS.md-style context file injection, resident in ``praisonai-code``.
|
||||
|
||||
This is the dependency-free home of the subtree-instruction helper so the
|
||||
terminal-native CLI (``pip install praisonai-code``) can load project
|
||||
instructions and resolve ``--instructions`` sources without importing the
|
||||
optional ``praisonai_bot`` support package or the umbrella ``praisonai``
|
||||
wrapper (C7 standalone contract).
|
||||
|
||||
It relies only on the standard library plus ``praisonaiagents.hooks`` (already a
|
||||
core dependency of ``praisonai-code``). The ``praisonai_bot`` copy remains a
|
||||
backward-compatible mirror for host apps that import it directly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import glob as _glob
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, List, Optional, Set
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_CANDIDATES = ["AGENTS.md", "agents.md", ".agents/AGENTS.md", "CLAUDE.md"]
|
||||
|
||||
# Upper bound on the bytes read from a single remote instruction source. Keeps a
|
||||
# stray large URL from ballooning the system context; oversized bodies are
|
||||
# truncated with a marker rather than failing the run.
|
||||
_REMOTE_MAX_BYTES = 256 * 1024
|
||||
|
||||
# Timeout (seconds) for a single remote instruction fetch. Best-effort: a slow
|
||||
# or unreachable URL is skipped with a warning, never blocking the run.
|
||||
_REMOTE_TIMEOUT = 5.0
|
||||
|
||||
# Opt-in to allow fetching instruction URLs that resolve to private, loopback,
|
||||
# or link-local addresses. Off by default so an auto-loaded project config from
|
||||
# an untrusted checkout cannot make the host contact internal services (SSRF).
|
||||
_ALLOW_LOCAL_URL_ENV = "PRAISONAI_INSTRUCTIONS_ALLOW_LOCAL_URLS"
|
||||
|
||||
# Tool-argument keys that carry a file path across the various file tools
|
||||
# (praisonaiagents ``read_file`` uses ``filepath``; praisonai-code tools use
|
||||
# ``path``/``file_path``; ACP edit tools use ``file_path``). The first present,
|
||||
# string-valued key wins.
|
||||
_PATH_ARG_KEYS = ("filepath", "file_path", "path", "filename", "target_file")
|
||||
|
||||
|
||||
def _get_git_root(start: Path) -> Optional[Path]:
|
||||
"""Resolve the git/project root using the resident CLI resolver.
|
||||
|
||||
Falls back to None when the helper or git is unavailable so discovery
|
||||
simply walks up to the filesystem root.
|
||||
"""
|
||||
try:
|
||||
from praisonai_code.cli.utils import project
|
||||
|
||||
return project.get_git_root(str(start))
|
||||
except (ImportError, OSError):
|
||||
return None
|
||||
|
||||
|
||||
# When no git root is found, cap the walk-up to avoid scanning arbitrary
|
||||
# system directories up to the filesystem root.
|
||||
_MAX_WALK_UP_DEPTH = 10
|
||||
|
||||
|
||||
def _discover_search_dirs(base: Path, walk_up: bool) -> List[Path]:
|
||||
"""Build directories to search, ordered root -> cwd (nearest last).
|
||||
|
||||
Walks up from ``base`` to the git root so nearer, more specific
|
||||
instruction files take precedence by appearing last. When no git root
|
||||
is found, the walk-up is capped at ``_MAX_WALK_UP_DEPTH`` levels to avoid
|
||||
scanning arbitrary system directories.
|
||||
"""
|
||||
base = base.resolve()
|
||||
if not walk_up:
|
||||
return [base]
|
||||
|
||||
git_root = _get_git_root(base)
|
||||
if git_root is not None:
|
||||
git_root = git_root.resolve()
|
||||
|
||||
dirs: List[Path] = []
|
||||
current = base
|
||||
depth = 0
|
||||
while True:
|
||||
dirs.append(current)
|
||||
if git_root and current == git_root:
|
||||
break
|
||||
if current == current.parent:
|
||||
break
|
||||
if git_root is None and depth >= _MAX_WALK_UP_DEPTH:
|
||||
break
|
||||
current = current.parent
|
||||
depth += 1
|
||||
|
||||
# Reverse so root is first and cwd (nearest) is last.
|
||||
dirs.reverse()
|
||||
return dirs
|
||||
|
||||
|
||||
def load_context_files(
|
||||
paths: Optional[List[str]] = None,
|
||||
*,
|
||||
cwd: Optional[Path] = None,
|
||||
walk_up: bool = True,
|
||||
) -> str:
|
||||
"""Load context from AGENTS.md-style files and return combined text.
|
||||
|
||||
Discovery mirrors the configuration resolver: it walks up from ``cwd``
|
||||
to the project boundary (git root) collecting instruction files at each
|
||||
level, layers a user-global file (``~/.praisonai/AGENTS.md``) as the
|
||||
lowest-precedence source, and concatenates root -> cwd so nearer files
|
||||
take precedence (appear last).
|
||||
|
||||
Args:
|
||||
paths: Explicit relative file names to load from ``cwd`` only. When
|
||||
provided, discovery/walk-up is skipped (backward compatible).
|
||||
cwd: Directory to start discovery from (defaults to ``Path.cwd()``).
|
||||
walk_up: When True (default), walk up to the git/project root and
|
||||
layer files. Ignored when ``paths`` is given.
|
||||
|
||||
Returns:
|
||||
Combined instruction text, blank-line separated.
|
||||
"""
|
||||
base = cwd or Path.cwd()
|
||||
|
||||
seen: set = set()
|
||||
chunks: List[str] = []
|
||||
|
||||
def _add(path: Path) -> None:
|
||||
if not path.is_file():
|
||||
return
|
||||
# De-duplicate by filesystem identity so the same physical file is
|
||||
# read once even via different paths or on case-insensitive volumes.
|
||||
try:
|
||||
stat = path.stat()
|
||||
key = (stat.st_dev, stat.st_ino)
|
||||
except OSError:
|
||||
key = path.resolve()
|
||||
if key in seen:
|
||||
return
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return
|
||||
seen.add(key)
|
||||
chunks.append(text)
|
||||
|
||||
# Explicit paths override discovery and remain cwd-only for compatibility.
|
||||
if paths is not None:
|
||||
for name in paths:
|
||||
_add(base / name)
|
||||
return "\n\n".join(chunks)
|
||||
|
||||
# Lowest-precedence layer: user-global instructions
|
||||
# (skipped when walk_up is False to honour cwd-only semantics).
|
||||
if walk_up:
|
||||
_add(Path.home() / ".praisonai" / "AGENTS.md")
|
||||
|
||||
# Walk-up layers: root -> cwd so nearer files take precedence (last).
|
||||
for search_dir in _discover_search_dirs(base, walk_up):
|
||||
for name in DEFAULT_CANDIDATES:
|
||||
_add(search_dir / name)
|
||||
|
||||
return "\n\n".join(chunks)
|
||||
|
||||
|
||||
def _is_remote_source(entry: str) -> bool:
|
||||
"""Whether an instruction entry is an ``http(s)://`` URL."""
|
||||
return entry.startswith(("http://", "https://"))
|
||||
|
||||
|
||||
def _allow_local_urls() -> bool:
|
||||
"""Whether fetching URLs that resolve to internal addresses is opted in."""
|
||||
return os.environ.get(_ALLOW_LOCAL_URL_ENV, "").strip().lower() in (
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
)
|
||||
|
||||
|
||||
def _is_blocked_host(host: str) -> bool:
|
||||
"""Whether ``host`` resolves to a private/loopback/link-local/reserved IP.
|
||||
|
||||
SSRF guard: an instruction URL can arrive from an auto-loaded project config
|
||||
in an untrusted checkout, so a URL that resolves to an internal service must
|
||||
not be fetched. Any resolved address in a non-global range blocks the fetch
|
||||
(fail-closed on resolution errors). Bypassable via ``_ALLOW_LOCAL_URL_ENV``
|
||||
for trusted internal setups.
|
||||
"""
|
||||
import ipaddress
|
||||
import socket
|
||||
|
||||
if not host:
|
||||
return True
|
||||
try:
|
||||
infos = socket.getaddrinfo(host, None)
|
||||
except OSError:
|
||||
# Unresolvable — let urlopen surface the failure/warning downstream.
|
||||
return False
|
||||
for info in infos:
|
||||
addr = info[4][0]
|
||||
try:
|
||||
ip = ipaddress.ip_address(addr.split("%", 1)[0])
|
||||
except ValueError:
|
||||
return True
|
||||
if (
|
||||
ip.is_private
|
||||
or ip.is_loopback
|
||||
or ip.is_link_local
|
||||
or ip.is_reserved
|
||||
or ip.is_multicast
|
||||
or ip.is_unspecified
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _url_is_fetchable(url: str) -> bool:
|
||||
"""Validate scheme and destination of a remote instruction URL (SSRF guard)."""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
if _allow_local_urls():
|
||||
return True
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme not in ("http", "https"):
|
||||
return False
|
||||
if _is_blocked_host(parsed.hostname or ""):
|
||||
logger.warning(
|
||||
"Skipping remote instruction source %s: resolves to a "
|
||||
"private/loopback/link-local address (set %s=1 to allow)",
|
||||
url,
|
||||
_ALLOW_LOCAL_URL_ENV,
|
||||
)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _fetch_remote_source(url: str) -> Optional[str]:
|
||||
"""Fetch a remote instruction source, best-effort and size-bounded.
|
||||
|
||||
Uses the stdlib ``urllib`` (lazily imported) so no heavy dependency is
|
||||
added. Destinations are validated against an SSRF guard (see
|
||||
:func:`_url_is_fetchable`) that rejects private/loopback/link-local hosts,
|
||||
and redirects are re-validated so a public URL cannot bounce to an internal
|
||||
one. A slow, unreachable, or oversized response is handled gracefully:
|
||||
failures return ``None`` (with a warning) so the run continues, and bodies
|
||||
larger than ``_REMOTE_MAX_BYTES`` are truncated with a marker.
|
||||
"""
|
||||
if not _url_is_fetchable(url):
|
||||
return None
|
||||
try:
|
||||
import urllib.request
|
||||
|
||||
class _GuardedRedirectHandler(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||||
if not _url_is_fetchable(newurl):
|
||||
return None
|
||||
return super().redirect_request(
|
||||
req, fp, code, msg, headers, newurl
|
||||
)
|
||||
|
||||
opener = urllib.request.build_opener(_GuardedRedirectHandler())
|
||||
with opener.open(url, timeout=_REMOTE_TIMEOUT) as resp: # nosec B310
|
||||
raw = resp.read(_REMOTE_MAX_BYTES + 1)
|
||||
except Exception as exc: # noqa: BLE001 - best-effort; never block the run
|
||||
logger.warning("Skipping remote instruction source %s: %s", url, exc)
|
||||
return None
|
||||
|
||||
truncated = len(raw) > _REMOTE_MAX_BYTES
|
||||
if truncated:
|
||||
raw = raw[:_REMOTE_MAX_BYTES]
|
||||
try:
|
||||
text = raw.decode("utf-8", errors="replace")
|
||||
except Exception: # noqa: BLE001 - decode is defensive
|
||||
return None
|
||||
if truncated:
|
||||
text += "\n... [remote instruction source truncated]"
|
||||
return text
|
||||
|
||||
|
||||
def resolve_instruction_sources(
|
||||
entries: Optional[List[str]] = None,
|
||||
*,
|
||||
cwd: Optional[Path] = None,
|
||||
) -> str:
|
||||
"""Resolve config-declared instruction sources into combined text.
|
||||
|
||||
Each entry may be:
|
||||
|
||||
* a plain file path (``docs/rules.md``),
|
||||
* a glob (``docs/standards/*.md``),
|
||||
* a ``~``-prefixed path (``~/company/ai-rules.md``), or
|
||||
* a remote ``http(s)://`` URL.
|
||||
|
||||
Entries are resolved in order and their contents concatenated (blank-line
|
||||
separated), so callers can layer org-wide sources before project-specific
|
||||
ones. Local globs expand to their matches sorted for determinism; a
|
||||
``~``/env-var path is expanded; remote URLs are fetched lazily, best-effort
|
||||
and size-bounded (see :func:`_fetch_remote_source`). Missing local paths and
|
||||
failed fetches are skipped with a warning rather than aborting the run.
|
||||
|
||||
Args:
|
||||
entries: Ordered list of source specifiers. ``None``/empty returns "".
|
||||
cwd: Base directory for resolving relative paths (defaults to CWD).
|
||||
|
||||
Returns:
|
||||
Combined instruction text, blank-line separated (possibly empty).
|
||||
"""
|
||||
if not entries:
|
||||
return ""
|
||||
|
||||
base = Path(cwd) if cwd else Path.cwd()
|
||||
|
||||
seen: Set = set()
|
||||
chunks: List[str] = []
|
||||
|
||||
def _add_file(path: Path) -> None:
|
||||
if not path.is_file():
|
||||
return
|
||||
try:
|
||||
stat = path.stat()
|
||||
key = (stat.st_dev, stat.st_ino)
|
||||
except OSError:
|
||||
key = path.resolve()
|
||||
if key in seen:
|
||||
return
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
logger.warning("Skipping instruction source %s: %s", path, exc)
|
||||
return
|
||||
seen.add(key)
|
||||
chunks.append(text)
|
||||
|
||||
for entry in entries:
|
||||
if not isinstance(entry, str) or not entry.strip():
|
||||
continue
|
||||
entry = entry.strip()
|
||||
|
||||
if _is_remote_source(entry):
|
||||
if entry in seen:
|
||||
continue
|
||||
text = _fetch_remote_source(entry)
|
||||
if text is not None:
|
||||
seen.add(entry)
|
||||
chunks.append(text)
|
||||
continue
|
||||
|
||||
expanded = os.path.expanduser(os.path.expandvars(entry))
|
||||
candidate = Path(expanded)
|
||||
if not candidate.is_absolute():
|
||||
candidate = base / candidate
|
||||
|
||||
# Expand globs (including brace-free ``*``/``?``/``[]`` patterns). When
|
||||
# the entry contains no glob metacharacter this yields the single path.
|
||||
pattern = str(candidate)
|
||||
if any(ch in pattern for ch in "*?["):
|
||||
for match in sorted(_glob.glob(pattern, recursive=True)):
|
||||
_add_file(Path(match))
|
||||
else:
|
||||
if candidate.is_file():
|
||||
_add_file(candidate)
|
||||
else:
|
||||
logger.warning("Instruction source not found: %s", entry)
|
||||
|
||||
return "\n\n".join(chunks)
|
||||
|
||||
|
||||
def _identity_key(path: Path):
|
||||
"""Filesystem identity for dedup (device+inode), falling back to resolved path."""
|
||||
try:
|
||||
stat = path.stat()
|
||||
return (stat.st_dev, stat.st_ino)
|
||||
except OSError:
|
||||
return path.resolve()
|
||||
|
||||
|
||||
class PathContextAttacher:
|
||||
"""Attach the nearest governing instruction files as files are touched.
|
||||
|
||||
Up-front ``load_context_files`` discovery walks ``cwd -> project root`` once
|
||||
at session start. In a monorepo, an agent that later reads/edits a file in a
|
||||
*sibling or deeper* subtree (e.g. ``packages/foo/AGENTS.md``) never sees that
|
||||
subtree's conventions. This attacher augments the up-front load: on the first
|
||||
touch of a directory it walks up to the project root collecting
|
||||
``DEFAULT_CANDIDATES``, deduplicates against files already loaded (both the
|
||||
up-front rules and earlier touches), bounds the result by a character budget,
|
||||
and caches per directory so repeated touches never re-walk disk.
|
||||
|
||||
Session-scoped: create one instance per session/agent run so cache and dedup
|
||||
state stay isolated (multi-agent safe).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
already_loaded: Optional[str] = None,
|
||||
*,
|
||||
max_chars: int = 8000,
|
||||
) -> None:
|
||||
"""Initialise the attacher.
|
||||
|
||||
Args:
|
||||
already_loaded: Text already injected up front (used only to seed
|
||||
dedup so its files are not re-attached). Its identity is tracked
|
||||
by content so duplicate physical files are skipped.
|
||||
max_chars: Character budget for the total text this attacher emits
|
||||
across the session. ``0`` disables the budget.
|
||||
"""
|
||||
self._seen: Set = set()
|
||||
self._seen_texts: Set[str] = set()
|
||||
self._dir_cache: Dict[Path, str] = {}
|
||||
self._max_chars = max_chars
|
||||
self._emitted_chars = 0
|
||||
# Seed dedup with the identities of already-loaded files so the same
|
||||
# physical instruction file discovered up front is not re-attached.
|
||||
if already_loaded:
|
||||
self._seed_from_text(already_loaded)
|
||||
|
||||
def _seed_from_text(self, text: str) -> None:
|
||||
# We cannot recover file identities from text alone, so we record the
|
||||
# text content to skip re-emitting identical bodies. Physical-identity
|
||||
# dedup below still handles the same file reached via different paths.
|
||||
#
|
||||
# ``load_context_files`` joins multiple instruction files with "\n\n",
|
||||
# so seed the whole blob *and* each chunk. This lets an individual file
|
||||
# discovered later match even when it was originally injected inside a
|
||||
# larger concatenated up-front load.
|
||||
self._seen_texts.add(text)
|
||||
for chunk in text.split("\n\n"):
|
||||
if chunk:
|
||||
self._seen_texts.add(chunk)
|
||||
|
||||
def attach_for_path(self, file_path) -> str:
|
||||
"""Return nearest instruction text for ``file_path``'s directory.
|
||||
|
||||
Walks up from the file's directory to the project root collecting
|
||||
instruction files (root first, nearest last), deduplicated against
|
||||
everything already emitted/loaded and bounded by the char budget. The
|
||||
first touch of a directory does the disk walk; later touches of the same
|
||||
directory return the cached (already-deduped) result cheaply.
|
||||
|
||||
Returns an empty string when nothing new is found or the budget is
|
||||
exhausted.
|
||||
"""
|
||||
p = Path(file_path)
|
||||
directory = (p if p.is_dir() else p.parent).resolve()
|
||||
|
||||
if directory in self._dir_cache:
|
||||
return self._dir_cache[directory]
|
||||
|
||||
# Collect newly-discovered files without mutating dedup state yet: if
|
||||
# the char budget discards this result we must not permanently lock the
|
||||
# files out of future touches (they were never actually emitted).
|
||||
chunks: List[str] = []
|
||||
new_keys: List = []
|
||||
local_seen_texts: Set[str] = set()
|
||||
for search_dir in _discover_search_dirs(directory, walk_up=True):
|
||||
for name in DEFAULT_CANDIDATES:
|
||||
candidate = search_dir / name
|
||||
if not candidate.is_file():
|
||||
continue
|
||||
key = _identity_key(candidate)
|
||||
if key in self._seen:
|
||||
continue
|
||||
try:
|
||||
text = candidate.read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
continue
|
||||
if text in self._seen_texts or text in local_seen_texts:
|
||||
self._seen.add(key)
|
||||
continue
|
||||
local_seen_texts.add(text)
|
||||
new_keys.append(key)
|
||||
chunks.append(text)
|
||||
|
||||
result = "\n\n".join(chunks)
|
||||
|
||||
# Enforce the character budget across the whole session, keeping the
|
||||
# truncation marker inside the remaining budget so max_chars holds.
|
||||
if self._max_chars and result:
|
||||
remaining = self._max_chars - self._emitted_chars
|
||||
if remaining <= 0:
|
||||
result = ""
|
||||
elif len(result) > remaining:
|
||||
suffix = "\n... [subtree context truncated]"
|
||||
if remaining > len(suffix):
|
||||
result = result[: remaining - len(suffix)] + suffix
|
||||
else:
|
||||
result = result[:remaining]
|
||||
|
||||
# Only commit dedup state when something was actually emitted, so a
|
||||
# budget-suppressed discovery can still surface if budget frees up.
|
||||
if result:
|
||||
for key in new_keys:
|
||||
self._seen.add(key)
|
||||
self._seen_texts |= local_seen_texts
|
||||
|
||||
self._emitted_chars += len(result)
|
||||
self._dir_cache[directory] = result
|
||||
return result
|
||||
|
||||
|
||||
def load_context_files_for_path(
|
||||
file_path,
|
||||
*,
|
||||
already_loaded: Optional[str] = None,
|
||||
max_chars: int = 8000,
|
||||
) -> str:
|
||||
"""Stateless one-shot discovery of nearest instruction files for a path.
|
||||
|
||||
Convenience wrapper around :class:`PathContextAttacher` for callers that do
|
||||
not maintain session state. Prefer the class when touching many files so
|
||||
dedup and per-directory caching persist across touches.
|
||||
"""
|
||||
return PathContextAttacher(
|
||||
already_loaded, max_chars=max_chars
|
||||
).attach_for_path(file_path)
|
||||
|
||||
|
||||
# Tool names that operate on a file and should trigger subtree-rule discovery.
|
||||
# Covers the real interactive tool surface (praisonaiagents ``read_file``/
|
||||
# ``write_file`` and the ACP ``acp_create_file``/``acp_edit_file``/
|
||||
# ``acp_delete_file`` tools) plus common generic aliases so custom file tools
|
||||
# are matched too.
|
||||
_FILE_TOOL_MATCHER = (
|
||||
r"^(read_file|write_file|edit_file|list_files"
|
||||
r"|acp_create_file|acp_edit_file|acp_delete_file"
|
||||
r"|read|edit|write|apply_patch|str_replace|multi_edit|search_replace)$"
|
||||
)
|
||||
|
||||
|
||||
def _extract_tool_path(tool_input: Any) -> Optional[str]:
|
||||
"""Pull the target file path out of a tool's input arguments.
|
||||
|
||||
Returns the first present, non-empty string value among ``_PATH_ARG_KEYS``
|
||||
so it works across the differing file-tool argument conventions.
|
||||
"""
|
||||
if not isinstance(tool_input, dict):
|
||||
return None
|
||||
for key in _PATH_ARG_KEYS:
|
||||
value = tool_input.get(key)
|
||||
if isinstance(value, str) and value:
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def build_subtree_context_hook(
|
||||
already_loaded: Optional[str] = None,
|
||||
*,
|
||||
max_chars: int = 8000,
|
||||
) -> Callable[[Any], Any]:
|
||||
"""Build an ``AFTER_TOOL`` hook that lazily attaches subtree instructions.
|
||||
|
||||
Wires the :class:`PathContextAttacher` into the neutral core hook
|
||||
substrate: on each file-touching tool call (``read_file``/``edit_file``/
|
||||
``write_file`` and friends) it resolves the nearest instruction file for the
|
||||
touched directory and, if not already loaded/emitted this session, returns it
|
||||
as ``additional_context`` so the agent sees locally-relevant rules the first
|
||||
time it works in that subtree.
|
||||
|
||||
The returned callable is session-scoped (its attacher owns the dedup and
|
||||
per-directory cache), so create one hook per session/agent run. Register it
|
||||
on an agent's hook registry for ``HookEvent.AFTER_TOOL`` (optionally with
|
||||
``matcher=file_tool_matcher()`` to skip non-file tools cheaply).
|
||||
|
||||
Args:
|
||||
already_loaded: Up-front instruction text (used only to seed dedup so
|
||||
files already injected at session start are not re-attached).
|
||||
max_chars: Character budget for the total text emitted across the
|
||||
session. ``0`` disables the budget.
|
||||
|
||||
Returns:
|
||||
A function ``hook(hook_input) -> HookResult | None`` suitable for
|
||||
``HookRegistry.register_function``. Returns ``None`` (no-op) when the
|
||||
tool is not a file tool or nothing new is discovered.
|
||||
"""
|
||||
attacher = PathContextAttacher(already_loaded, max_chars=max_chars)
|
||||
|
||||
def _hook(hook_input: Any):
|
||||
tool_input = getattr(hook_input, "tool_input", None)
|
||||
path = _extract_tool_path(tool_input)
|
||||
if not path:
|
||||
return None
|
||||
try:
|
||||
context = attacher.attach_for_path(path)
|
||||
except OSError:
|
||||
return None
|
||||
if not context:
|
||||
return None
|
||||
try:
|
||||
from praisonaiagents.hooks import HookResult
|
||||
except ImportError:
|
||||
return None
|
||||
return HookResult(decision="allow", additional_context=context)
|
||||
|
||||
_hook.__name__ = "subtree_instruction_injection"
|
||||
return _hook
|
||||
|
||||
|
||||
def file_tool_matcher() -> str:
|
||||
"""Regex matching file-touching tool names for the subtree-context hook."""
|
||||
return _FILE_TOOL_MATCHER
|
||||
@@ -0,0 +1,183 @@
|
||||
"""Standalone regression tests for project instructions & rules (issue #4337).
|
||||
|
||||
On a lean ``pip install praisonai-code`` (no ``praisonai`` wrapper, no
|
||||
``praisonai_bot`` support package) the terminal product must still:
|
||||
|
||||
* attach subtree ``AGENTS.md`` on cross-directory file reads,
|
||||
* resolve ``--instructions`` globs/URLs up front, and
|
||||
* run ``praisonai rules list/add/clear`` natively.
|
||||
|
||||
These tests simulate the absence of the optional packages by blocking their
|
||||
import so the resident ``praisonai_code`` implementation is exercised.
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
# Import the command modules up front (with the optional packages available) so
|
||||
# the ``no_optional_packages`` fixture only affects the *runtime* helper lookup
|
||||
# inside ``_wire_subtree_context_hook`` / the rules handler, not their import.
|
||||
from praisonai_code.cli.commands.run import _wire_subtree_context_hook
|
||||
from praisonai_code.cli.commands import rules as rules_cmd
|
||||
from praisonai_code.integration import context_files as _resident_ctx # noqa: F401
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_optional_packages(monkeypatch):
|
||||
"""Make ``praisonai_bot`` and the ``praisonai`` wrapper un-importable.
|
||||
|
||||
Simulates a standalone ``praisonai-code`` install so the resident helper
|
||||
(``praisonai_code.integration.context_files``) is the only one available at
|
||||
runtime. ``praisonai_code`` itself is a distinct top-level package and is
|
||||
unaffected.
|
||||
"""
|
||||
import importlib
|
||||
import sys
|
||||
|
||||
for name in list(sys.modules):
|
||||
if name == "praisonai_bot" or name.startswith("praisonai_bot."):
|
||||
monkeypatch.delitem(sys.modules, name, raising=False)
|
||||
if name == "praisonai" or name.startswith("praisonai."):
|
||||
monkeypatch.delitem(sys.modules, name, raising=False)
|
||||
|
||||
real_import_module = importlib.import_module
|
||||
|
||||
def _blocked_import_module(name, *args, **kwargs):
|
||||
if name == "praisonai_bot" or name.startswith("praisonai_bot."):
|
||||
raise ImportError(f"blocked in standalone test: {name}")
|
||||
if name == "praisonai" or name.startswith("praisonai."):
|
||||
raise ImportError(f"blocked in standalone test: {name}")
|
||||
return real_import_module(name, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(importlib, "import_module", _blocked_import_module)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def monorepo(tmp_path, monkeypatch):
|
||||
"""A monorepo with a root and a nested package, with a fake git root."""
|
||||
root = tmp_path / "repo"
|
||||
foo = root / "packages" / "foo"
|
||||
foo.mkdir(parents=True)
|
||||
(root / "AGENTS.md").write_text("ROOT RULES")
|
||||
(foo / "AGENTS.md").write_text("FOO RULES")
|
||||
# Force deterministic git-root detection on the *resident* helper.
|
||||
monkeypatch.setattr(
|
||||
"praisonai_code.integration.context_files._get_git_root",
|
||||
lambda start: root,
|
||||
)
|
||||
return root, foo
|
||||
|
||||
|
||||
def _fire_after_tool(registry, tool_input):
|
||||
from praisonaiagents.hooks import HookEvent
|
||||
|
||||
event = SimpleNamespace(tool_input=tool_input)
|
||||
contexts = []
|
||||
for hook in registry.get_hooks(HookEvent.AFTER_TOOL):
|
||||
func = getattr(hook, "func", hook)
|
||||
result = func(event)
|
||||
if result is not None and getattr(result, "additional_context", None):
|
||||
contexts.append(result.additional_context)
|
||||
return "\n".join(contexts)
|
||||
|
||||
|
||||
def test_subtree_hook_wired_without_optional_packages(
|
||||
monorepo, no_optional_packages
|
||||
):
|
||||
"""Subtree AGENTS.md attaches on cross-dir reads with no wrapper/bot."""
|
||||
_root, foo = monorepo
|
||||
agent_config = {"name": "RunAgent"}
|
||||
|
||||
_wire_subtree_context_hook(agent_config)
|
||||
|
||||
registry = agent_config.get("hooks")
|
||||
assert registry is not None, "resident subtree hook must register standalone"
|
||||
|
||||
context = _fire_after_tool(registry, {"file_path": str(foo / "bar.py")})
|
||||
assert "FOO RULES" in context
|
||||
|
||||
|
||||
def test_instruction_sources_resolved_without_optional_packages(
|
||||
tmp_path, monkeypatch, no_optional_packages
|
||||
):
|
||||
"""``--instructions`` globs resolve natively on a standalone install."""
|
||||
monkeypatch.setattr(
|
||||
"praisonai_code.integration.context_files._get_git_root",
|
||||
lambda start: tmp_path,
|
||||
)
|
||||
(tmp_path / "a.md").write_text("ORG STANDARD A")
|
||||
(tmp_path / "b.md").write_text("ORG STANDARD B")
|
||||
|
||||
agent_config = {"name": "RunAgent"}
|
||||
_wire_subtree_context_hook(
|
||||
agent_config, instructions=[str(tmp_path / "*.md")]
|
||||
)
|
||||
|
||||
backstory = agent_config.get("backstory") or ""
|
||||
assert "ORG STANDARD A" in backstory
|
||||
assert "ORG STANDARD B" in backstory
|
||||
assert "# Project Instructions" in backstory
|
||||
|
||||
|
||||
def test_resident_context_files_is_dependency_free(no_optional_packages):
|
||||
"""The resident helper imports and exposes the expected API standalone."""
|
||||
from praisonai_code.integration import context_files
|
||||
|
||||
assert callable(context_files.resolve_instruction_sources)
|
||||
assert callable(context_files.build_subtree_context_hook)
|
||||
assert callable(context_files.file_tool_matcher)
|
||||
|
||||
|
||||
# --- Native rules command --------------------------------------------------
|
||||
|
||||
|
||||
def test_rules_list_native_without_wrapper(tmp_path, monkeypatch, capsys):
|
||||
"""``rules list`` uses core RulesManager and never touches the wrapper."""
|
||||
(tmp_path / "AGENTS.md").write_text("be concise")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
def _boom(*_a, **_k):
|
||||
raise AssertionError("wrapper fallback must not be used standalone")
|
||||
|
||||
monkeypatch.setattr(rules_cmd, "_fallback", _boom)
|
||||
|
||||
rules_cmd.rules_list()
|
||||
out = capsys.readouterr().out
|
||||
# AGENTS.md is discovered as a root instruction rule.
|
||||
assert "AGENTS" in out or "agents" in out.lower()
|
||||
|
||||
|
||||
def test_rules_add_then_list_native(tmp_path, monkeypatch, capsys):
|
||||
"""``rules add`` creates a workspace rule surfaced by ``rules list``."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
monkeypatch.setattr(
|
||||
rules_cmd,
|
||||
"_fallback",
|
||||
lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("no fallback")),
|
||||
)
|
||||
|
||||
rules_cmd.rules_add("style", "use tabs")
|
||||
add_out = capsys.readouterr().out
|
||||
assert "style" in add_out
|
||||
|
||||
rules_cmd.rules_list()
|
||||
list_out = capsys.readouterr().out
|
||||
assert "style" in list_out
|
||||
|
||||
|
||||
def test_rules_clear_native(tmp_path, monkeypatch, capsys):
|
||||
"""``rules clear`` removes CLI-created rules."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
monkeypatch.setattr(
|
||||
rules_cmd,
|
||||
"_fallback",
|
||||
lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("no fallback")),
|
||||
)
|
||||
|
||||
rules_cmd.rules_add("style", "use tabs")
|
||||
capsys.readouterr()
|
||||
|
||||
rules_cmd.rules_clear()
|
||||
clear_out = capsys.readouterr().out
|
||||
assert "Cleared" in clear_out
|
||||
Reference in new issue
Block a user