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:
praisonai-triage-agent[bot]andMervinPraison committed 2026-08-25 08:59:51 +00:00
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