diff --git a/src/praisonai-agents/praisonaiagents/agent/agent.py b/src/praisonai-agents/praisonaiagents/agent/agent.py
index 2160d79e8..9501e2dab 100644
--- a/src/praisonai-agents/praisonaiagents/agent/agent.py
+++ b/src/praisonai-agents/praisonaiagents/agent/agent.py
@@ -2257,9 +2257,11 @@ def _add_skill_tools(self):
Uses lazy imports from praisonaiagents.tools to avoid performance impact
when skills are not used.
- Honours ``PRAISONAI_DISABLE_SKILL_TOOLS=1`` so hosts that do not want
- the subprocess-backed ``run_skill_script`` tool auto-injected can opt
- out without touching code.
+ G-E fix: run_skill_script is now safer by default:
+ - PRAISONAI_DISABLE_SKILL_TOOLS=1: disables all skill tools (explicit deny wins)
+ - PRAISONAI_ENABLE_SKILL_TOOLS=1: enables run_skill_script by default
+ - Any loaded skill with 'run_skill_script' in allowed-tools: enables it
+ - Otherwise: only read_file is added (safer default)
"""
import os as _os
if _os.environ.get("PRAISONAI_DISABLE_SKILL_TOOLS") in ("1", "true", "True"):
@@ -2274,7 +2276,7 @@ def _add_skill_tools(self):
elif hasattr(tool, 'name'):
tool_names.add(tool.name)
- # Add read_file if not present
+ # Add read_file if not present (low risk, always enabled)
if 'read_file' not in tool_names:
try:
from ..tools import read_file
@@ -2283,8 +2285,26 @@ def _add_skill_tools(self):
except ImportError:
logging.warning("Could not import read_file tool for skills")
- # Add run_skill_script from skill_tools module
- if 'run_skill_script' not in tool_names:
+ # G-E fix: run_skill_script safer by default
+ # Only add if explicitly enabled OR any skill declares it in allowed-tools
+ should_add_script_tool = False
+
+ # Check explicit environment enable
+ if _os.environ.get("PRAISONAI_ENABLE_SKILL_TOOLS") in ("1", "true", "True"):
+ should_add_script_tool = True
+ logging.debug("run_skill_script enabled via PRAISONAI_ENABLE_SKILL_TOOLS")
+
+ # Check if any loaded skill declares it in allowed-tools
+ if self._skill_manager and not should_add_script_tool:
+ for skill in self._skill_manager.skills:
+ allowed_tools = self._skill_manager.get_allowed_tools(skill.properties.name)
+ if "run_skill_script" in allowed_tools:
+ should_add_script_tool = True
+ logging.debug(f"run_skill_script enabled by skill '{skill.properties.name}' allowed-tools")
+ break
+
+ # Add run_skill_script if conditions are met
+ if should_add_script_tool and 'run_skill_script' not in tool_names:
try:
from ..tools.skill_tools import create_skill_tools
# Create skill tools with current working directory
diff --git a/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py b/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py
index 7079cbe20..3ea81a0e0 100644
--- a/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py
+++ b/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py
@@ -1064,30 +1064,38 @@ def _resolve_skill_invocation(self, prompt):
rendered = mgr.invoke(name, raw_args=args)
if rendered is None:
return prompt
- # G6: Best-effort pre-approve any tools declared under
+ # G-A fix: Best-effort pre-approve any tools declared under
# `allowed-tools` in the skill frontmatter. Non-fatal on error.
try:
tool_names = mgr.get_allowed_tools(name)
if tool_names:
- from ..approval import get_approval_registry, AutoApproveBackend
+ from ..approval import get_approval_registry
registry = get_approval_registry()
- agent_name = getattr(self, "name", None)
- for _tn in tool_names:
- try:
- registry.set_backend(
- AutoApproveBackend(),
- agent_name=agent_name,
- tool_name=_tn,
- )
- except TypeError:
- # Older registry may not accept tool_name kwarg
- registry.set_backend(AutoApproveBackend(), agent_name=agent_name)
- except Exception as e: # pragma: no cover - approval is optional
- from .._logging import get_logger
- logger = get_logger(__name__)
- logger.warning(f"Approval system initialization failed for agent {agent_name}: {e}")
- # Don't raise - approval is optional, but log the failure for debugging
+ agent_name = getattr(self, "display_name", getattr(self, "name", None))
+ if agent_name: # Only approve if we have a stable agent identifier
+ for _tn in tool_names:
+ try:
+ registry.auto_approve_tool(_tn, agent_name=agent_name)
+ except Exception as exc: # pragma: no cover - approval is optional
+ logging.debug(
+ "Failed to auto-approve skill tool '%s' for skill '%s' on agent '%s': %s. "
+ "The skill will continue, but this tool may still require explicit approval.",
+ _tn,
+ name,
+ agent_name,
+ exc,
+ exc_info=True,
+ )
+ except Exception as exc: # pragma: no cover - approval is optional
+ logging.debug(
+ "Failed to resolve allowed tools for skill '%s' on agent '%s': %s. "
+ "The skill will continue without pre-approving tools.",
+ name,
+ getattr(self, "name", None),
+ exc,
+ exc_info=True,
+ )
return rendered
def chat(self, prompt: str, temperature: float = 1.0, tools: Optional[List[Any]] = None, output_json: Optional[Any] = None, output_pydantic: Optional[Any] = None, reasoning_steps: bool = False, stream: Optional[bool] = None, task_name: Optional[str] = None, task_description: Optional[str] = None, task_id: Optional[str] = None, config: Optional[Dict[str, Any]] = None, force_retrieval: bool = False, skip_retrieval: bool = False, attachments: Optional[List[str]] = None, tool_choice: Optional[str] = None) -> Optional[str]:
diff --git a/src/praisonai-agents/praisonaiagents/approval/registry.py b/src/praisonai-agents/praisonaiagents/approval/registry.py
index 4de62349f..a71b234c2 100644
--- a/src/praisonai-agents/praisonaiagents/approval/registry.py
+++ b/src/praisonai-agents/praisonaiagents/approval/registry.py
@@ -76,6 +76,9 @@ def __init__(self) -> None:
self._required_tools: Set[str] = set()
self._risk_levels: Dict[str, str] = {}
+ # Per-agent, per-tool auto-approval (G-A fix)
+ self._agent_tool_auto_approve: Dict[tuple[str, str], bool] = {}
+
# Context variables (per-coroutine / per-thread)
self._approved_context: contextvars.ContextVar[Set[str]] = contextvars.ContextVar(
"approved_context", default=set()
@@ -139,6 +142,20 @@ def is_required(self, tool_name: str) -> bool:
def get_risk_level(self, tool_name: str) -> Optional[str]:
return self._risk_levels.get(tool_name)
+ # ── Per-tool auto-approval (G-A fix) ─────────────────────────────────
+
+ def auto_approve_tool(self, tool_name: str, agent_name: str) -> None:
+ """Pre-approve a single tool for a specific agent."""
+ if not agent_name:
+ raise ValueError("Skill auto-approval requires a stable agent/session scope")
+ self._agent_tool_auto_approve[(agent_name, tool_name)] = True
+
+ def is_auto_approved(self, tool_name: str, agent_name: str) -> bool:
+ """Check if a tool is auto-approved for a specific agent."""
+ if not agent_name:
+ return False
+ return self._agent_tool_auto_approve.get((agent_name, tool_name), False)
+
# ── Context helpers ──────────────────────────────────────────────────
def mark_approved(self, tool_name: str) -> None:
@@ -189,6 +206,11 @@ def approve_sync(
if self.is_already_approved(tool_name):
return ApprovalDecision(approved=True, reason="Already approved in context")
+ # Check per-tool auto-approval (G-A fix)
+ if self.is_auto_approved(tool_name, agent_name):
+ self.mark_approved(tool_name)
+ return ApprovalDecision(approved=True, reason="Auto-approved (skill)", approver="skill")
+
# Env auto-approve
if self.is_env_auto_approve():
self.mark_approved(tool_name)
@@ -237,6 +259,11 @@ async def approve_async(
if self.is_already_approved(tool_name):
return ApprovalDecision(approved=True, reason="Already approved in context")
+ # Check per-tool auto-approval (G-A fix)
+ if self.is_auto_approved(tool_name, agent_name):
+ self.mark_approved(tool_name)
+ return ApprovalDecision(approved=True, reason="Auto-approved (skill)", approver="skill")
+
if self.is_env_auto_approve():
self.mark_approved(tool_name)
return ApprovalDecision(approved=True, reason="Auto-approved (env)", approver="env")
diff --git a/src/praisonai-agents/praisonaiagents/skills/__init__.py b/src/praisonai-agents/praisonaiagents/skills/__init__.py
index 53cbefd06..89b18fec8 100644
--- a/src/praisonai-agents/praisonaiagents/skills/__init__.py
+++ b/src/praisonai-agents/praisonaiagents/skills/__init__.py
@@ -48,6 +48,12 @@
"SkillSourceProtocol",
"SkillInvocationPolicyProtocol",
"SkillMutatorProtocol",
+ "SkillActivationProtocol",
+ # Events
+ "SkillDiscoveredEvent",
+ "SkillActivatedEvent",
+ # Budget
+ "SkillPromptBudget",
]
@@ -93,6 +99,18 @@ def __getattr__(name: str):
from .protocols import SkillSourceProtocol, SkillInvocationPolicyProtocol, SkillMutatorProtocol
return locals()[name]
+ if name == "SkillActivationProtocol":
+ from .activation import SkillActivationProtocol
+ return SkillActivationProtocol
+
+ if name in ("SkillDiscoveredEvent", "SkillActivatedEvent"):
+ from .events import SkillDiscoveredEvent, SkillActivatedEvent
+ return locals()[name]
+
+ if name == "SkillPromptBudget":
+ from .budget import SkillPromptBudget
+ return SkillPromptBudget
+
if name == "load_skill":
# Fixes G12: praisonai.capabilities.skills.skill_load import target.
# Returns a LoadedSkill (metadata + activated instructions) by name,
diff --git a/src/praisonai-agents/praisonaiagents/skills/activation.py b/src/praisonai-agents/praisonaiagents/skills/activation.py
new file mode 100644
index 000000000..deb160608
--- /dev/null
+++ b/src/praisonai-agents/praisonaiagents/skills/activation.py
@@ -0,0 +1,29 @@
+"""Agent Skills activation protocol for progressive disclosure."""
+
+from typing import Protocol, Optional
+
+
+class SkillActivationProtocol(Protocol):
+ """Protocol for progressive disclosure skill activation.
+
+ This protocol defines the interface for activating skills on demand,
+ supporting Claude Code-style progressive disclosure where only skill
+ descriptions are shown in the system prompt initially, with full
+ bodies loaded when needed.
+ """
+
+ def activate(self, name: str, arguments: str = "", session_id: Optional[str] = None) -> str:
+ """Activate a skill and return its rendered body.
+
+ Args:
+ name: Name of the skill to activate
+ arguments: Arguments to substitute in the skill body
+ session_id: Optional session identifier for context
+
+ Returns:
+ Rendered skill body with arguments substituted
+
+ Raises:
+ ValueError: If skill is not found or not user-invocable
+ """
+ ...
\ No newline at end of file
diff --git a/src/praisonai-agents/praisonaiagents/skills/budget.py b/src/praisonai-agents/praisonaiagents/skills/budget.py
new file mode 100644
index 000000000..bad8bf091
--- /dev/null
+++ b/src/praisonai-agents/praisonaiagents/skills/budget.py
@@ -0,0 +1,93 @@
+"""Agent Skills prompt budget management.
+
+Controls how many skills and how much content gets included in
+the system prompt to prevent unbounded growth with large skill libraries.
+"""
+
+import html
+from dataclasses import dataclass
+from typing import Literal, List
+from .models import SkillMetadata
+
+# Maximum description chars before truncation (from prompt.py)
+MAX_COMBINED_DESCRIPTION_CHARS = 1536
+
+
+def _truncate(text: str, limit: int) -> str:
+ """Truncate text to limit (same logic as prompt.py)."""
+ if text is None:
+ return ""
+ if len(text) <= limit:
+ return text
+ return text[: max(0, limit - 1)].rstrip() + "\u2026"
+
+
+def _estimate_skill_xml_chars(skill: SkillMetadata) -> int:
+ """Estimate the rendered XML character count for a skill."""
+ name = html.escape(skill.name)
+ description = html.escape(_truncate(skill.description, MAX_COMBINED_DESCRIPTION_CHARS))
+ location = html.escape(skill.location or "")
+
+ # Exact XML format from format_skill_for_prompt
+ xml_content = f"""
+ {name}
+ {description}
+ {location}
+ """
+
+ return len(xml_content)
+
+
+@dataclass(frozen=True)
+class SkillPromptBudget:
+ """Budget constraints for skills included in system prompts.
+
+ Prevents unbounded system prompt growth when agents have access
+ to large skill libraries.
+ """
+ max_chars: int = 4096
+ max_skills: int = 50
+ strategy: Literal["priority", "fifo", "alpha"] = "fifo"
+
+
+def apply_budget(skills: List[SkillMetadata], budget: SkillPromptBudget) -> tuple[List[SkillMetadata], bool]:
+ """Apply budget constraints to a list of skills.
+
+ Args:
+ skills: List of skill metadata to potentially include
+ budget: Budget constraints to apply
+
+ Returns:
+ Tuple of (filtered_skills, was_truncated)
+ """
+ if not skills:
+ return skills, False
+
+ # Apply ordering strategy first, before selecting the bounded subset
+ if budget.strategy == "alpha":
+ ordered_skills = sorted(skills, key=lambda s: s.name)
+ elif budget.strategy == "fifo":
+ ordered_skills = skills
+ else:
+ # priority strategy would require skill metadata to include priority field;
+ # for now, treat as fifo
+ ordered_skills = skills
+
+ # Apply skill count limit after ordering
+ limited_skills = ordered_skills[:budget.max_skills]
+ skill_count_truncated = len(limited_skills) < len(skills)
+
+ total_chars = 0
+ filtered_skills = []
+ char_truncated = False
+
+ for skill in limited_skills:
+ skill_chars = _estimate_skill_xml_chars(skill)
+ if total_chars + skill_chars > budget.max_chars:
+ char_truncated = True
+ break
+ filtered_skills.append(skill)
+ total_chars += skill_chars
+
+ was_truncated = skill_count_truncated or char_truncated
+ return filtered_skills, was_truncated
\ No newline at end of file
diff --git a/src/praisonai-agents/praisonaiagents/skills/events.py b/src/praisonai-agents/praisonaiagents/skills/events.py
new file mode 100644
index 000000000..b368eeb7c
--- /dev/null
+++ b/src/praisonai-agents/praisonaiagents/skills/events.py
@@ -0,0 +1,33 @@
+"""Agent Skills observability events for telemetry and monitoring."""
+
+from dataclasses import dataclass
+from typing import Literal
+
+
+@dataclass
+class SkillDiscoveredEvent:
+ """Event emitted when a skill is discovered during skill directory scanning.
+
+ This event provides visibility into which skills are found and from
+ what sources during the discovery phase.
+ """
+ agent: str
+ skill_name: str
+ source: str # Directory path or source identifier
+ description_chars: int # Length of skill description
+
+
+@dataclass
+class SkillActivatedEvent:
+ """Event emitted when a skill is activated (full body loaded/rendered).
+
+ This event tracks skill usage patterns and performance metrics
+ for skill invocation.
+ """
+ agent: str
+ skill_name: str
+ trigger: Literal["slash", "activate_tool", "auto"]
+ arguments: str
+ rendered_chars: int
+ session_id: str | None = None
+ activation_time_ms: float | None = None
\ No newline at end of file
diff --git a/src/praisonai-agents/praisonaiagents/skills/prompt.py b/src/praisonai-agents/praisonaiagents/skills/prompt.py
index 4c428c728..8d09b05b4 100644
--- a/src/praisonai-agents/praisonaiagents/skills/prompt.py
+++ b/src/praisonai-agents/praisonaiagents/skills/prompt.py
@@ -1,11 +1,12 @@
"""XML prompt generation for Agent Skills."""
from pathlib import Path
-from typing import List
+from typing import List, Optional
import html
from .models import SkillMetadata
from .parser import read_properties
+from .budget import SkillPromptBudget, apply_budget
# G22: Claude Code-equivalent cap on combined description + when_to_use.
@@ -37,12 +38,13 @@ def format_skill_for_prompt(skill: SkillMetadata) -> str:
"""
-def generate_skills_xml(skills: List[SkillMetadata], working_directory: str = None) -> str:
+def generate_skills_xml(skills: List[SkillMetadata], working_directory: str | None = None, budget: SkillPromptBudget | None = None) -> str:
"""Generate XML block for available skills.
Args:
skills: List of SkillMetadata instances
working_directory: Current working directory for path resolution
+ budget: Optional budget to limit skills included in prompt
Returns:
XML string with block
@@ -50,16 +52,23 @@ def generate_skills_xml(skills: List[SkillMetadata], working_directory: str = No
import os
cwd = working_directory or os.getcwd()
+ # Apply budget if provided
+ if budget and skills:
+ skills, was_truncated = apply_budget(skills, budget)
+ truncation_note = " Some skills were omitted due to prompt budget limits." if was_truncated else ""
+ else:
+ truncation_note = ""
+
if not skills:
return f"""
{html.escape(cwd)}
- When using run_skill_script, pass file paths relative to the working directory or as absolute paths. The tool will resolve relative paths automatically.
+ When using run_skill_script, pass file paths relative to the working directory or as absolute paths. The tool will resolve relative paths automatically.{truncation_note}
"""
skill_entries = "\n".join(format_skill_for_prompt(s) for s in skills)
return f"""
{html.escape(cwd)}
- When using run_skill_script, pass file paths relative to the working directory or as absolute paths. The tool will resolve relative paths automatically.
+ When using run_skill_script, pass file paths relative to the working directory or as absolute paths. The tool will resolve relative paths automatically.{truncation_note}
{skill_entries}
"""
diff --git a/src/praisonai-agents/tests/unit/skills/test_approval_scoping.py b/src/praisonai-agents/tests/unit/skills/test_approval_scoping.py
new file mode 100644
index 000000000..35255e239
--- /dev/null
+++ b/src/praisonai-agents/tests/unit/skills/test_approval_scoping.py
@@ -0,0 +1,97 @@
+"""
+Test G-A regression: approval scoping bug fix.
+
+Verifies that skills with allowed-tools do not leak approval across agents.
+"""
+
+import pytest
+from unittest.mock import Mock, patch
+from pathlib import Path
+import tempfile
+import os
+
+from praisonaiagents.approval import get_approval_registry
+
+
+def test_skill_allowed_tools_do_not_leak_across_agents():
+ """G-A regression test: skill allowed-tools should not leak across agents.
+
+ The pre-fix bug: ``_resolve_skill_invocation`` set ``AutoApproveBackend``
+ for the whole agent (via ``set_backend(...)``), which granted approval for
+ every tool on that agent. The fix pre-approves only the named tools via
+ ``registry.auto_approve_tool`` scoped to ``(agent_name, tool_name)``.
+ """
+ registry = get_approval_registry()
+ registry._agent_tool_auto_approve.clear()
+
+ # Mock the skill_manager to avoid constructing a real Agent.
+ agent_x = Mock()
+ agent_x.name = "agent_x"
+ agent_x.display_name = "agent_x"
+ agent_x.skill_manager = Mock()
+ agent_x.skill_manager.get_allowed_tools.return_value = ["read_file"]
+ agent_x.skill_manager.invoke.return_value = "Skill activated"
+
+ from praisonaiagents.agent.chat_mixin import ChatMixin
+ agent_x._resolve_skill_invocation = ChatMixin._resolve_skill_invocation.__get__(agent_x)
+
+ result = agent_x._resolve_skill_invocation("/demo")
+ assert result == "Skill activated"
+
+ # Correct per-agent, per-tool scoping:
+ assert registry.is_auto_approved("read_file", agent_name="agent_x")
+ assert not registry.is_auto_approved("read_file", agent_name="agent_y")
+ assert not registry.is_auto_approved("write_file", agent_name="agent_x")
+
+ # Pre-fix bug signature: agent-wide backend swap must NOT happen anymore.
+ assert registry._global_backend is None
+ assert "agent_x" not in registry._agent_backends
+
+
+def test_approval_registry_auto_approve_methods():
+ """``auto_approve_tool`` requires a stable per-agent scope (no global form)."""
+ registry = get_approval_registry()
+ registry._agent_tool_auto_approve.clear()
+
+ registry.auto_approve_tool("read_file", agent_name="agent1")
+
+ assert registry.is_auto_approved("read_file", agent_name="agent1")
+ assert not registry.is_auto_approved("read_file", agent_name="agent2")
+ assert not registry.is_auto_approved("write_file", agent_name="agent1")
+
+ # Explicit design: refuse anonymous / global scope for skill approvals.
+ with pytest.raises(ValueError):
+ registry.auto_approve_tool("write_file", agent_name=None)
+ with pytest.raises(ValueError):
+ registry.auto_approve_tool("write_file", agent_name="")
+
+
+def test_approval_decision_includes_auto_approved_tools():
+ """Sync approval honours per-agent auto-approval."""
+ registry = get_approval_registry()
+ registry._agent_tool_auto_approve.clear()
+ registry.clear_approved()
+
+ registry.add_requirement("read_file", "high")
+ registry.auto_approve_tool("read_file", agent_name="agent1")
+
+ decision = registry.approve_sync("agent1", "read_file", {})
+ assert decision.approved
+ assert decision.reason == "Auto-approved (skill)"
+ assert decision.approver == "skill"
+
+
+@pytest.mark.asyncio
+async def test_async_approval_decision_includes_auto_approved_tools():
+ """Async approval honours per-agent auto-approval."""
+ registry = get_approval_registry()
+ registry._agent_tool_auto_approve.clear()
+ registry.clear_approved() # context-var reset so "Already approved" doesn't fire
+
+ registry.add_requirement("read_file", "high")
+ registry.auto_approve_tool("read_file", agent_name="agent1")
+
+ decision = await registry.approve_async("agent1", "read_file", {})
+ assert decision.approved
+ assert decision.reason == "Auto-approved (skill)"
+ assert decision.approver == "skill"
\ No newline at end of file
diff --git a/src/praisonai-agents/tests/unit/skills/test_budget.py b/src/praisonai-agents/tests/unit/skills/test_budget.py
new file mode 100644
index 000000000..a0ea178c0
--- /dev/null
+++ b/src/praisonai-agents/tests/unit/skills/test_budget.py
@@ -0,0 +1,128 @@
+"""
+Tests for G-D: SkillPromptBudget functionality.
+"""
+
+import pytest
+from praisonaiagents.skills.budget import SkillPromptBudget, apply_budget
+from praisonaiagents.skills.models import SkillMetadata
+
+
+def create_test_skill(name: str, description: str) -> SkillMetadata:
+ """Helper to create test skill metadata."""
+ return SkillMetadata(
+ name=name,
+ description=description,
+ location=f"./skills/{name}"
+ )
+
+
+def test_skill_prompt_budget_defaults():
+ """Test SkillPromptBudget default values."""
+ budget = SkillPromptBudget()
+ assert budget.max_chars == 4096
+ assert budget.max_skills == 50
+ assert budget.strategy == "fifo"
+
+
+def test_apply_budget_no_skills():
+ """Test budget application with empty skill list."""
+ budget = SkillPromptBudget(max_chars=1000, max_skills=10)
+ skills, was_truncated = apply_budget([], budget)
+ assert skills == []
+ assert was_truncated is False
+
+
+def test_apply_budget_under_limits():
+ """Test budget application when skills are under limits."""
+ skills = [
+ create_test_skill("skill1", "Short description"),
+ create_test_skill("skill2", "Another short description"),
+ ]
+ budget = SkillPromptBudget(max_chars=1000, max_skills=10)
+
+ filtered_skills, was_truncated = apply_budget(skills, budget)
+ assert len(filtered_skills) == 2
+ assert was_truncated is False
+ assert filtered_skills[0].name == "skill1"
+ assert filtered_skills[1].name == "skill2"
+
+
+def test_apply_budget_skill_count_limit():
+ """Test budget application when skill count exceeds limit."""
+ skills = [
+ create_test_skill(f"skill{i}", "Description")
+ for i in range(1, 6) # 5 skills
+ ]
+ budget = SkillPromptBudget(max_chars=10000, max_skills=3)
+
+ filtered_skills, was_truncated = apply_budget(skills, budget)
+ assert len(filtered_skills) == 3
+ assert was_truncated is True
+ # Should keep first 3 skills
+ assert filtered_skills[0].name == "skill1"
+ assert filtered_skills[2].name == "skill3"
+
+
+def test_apply_budget_char_limit():
+ """Test budget application when character count exceeds limit."""
+ skills = [
+ create_test_skill("skill1", "A" * 100), # ~150 chars with XML
+ create_test_skill("skill2", "B" * 100), # ~150 chars with XML
+ create_test_skill("skill3", "C" * 100), # ~150 chars with XML
+ ]
+ budget = SkillPromptBudget(max_chars=250, max_skills=10) # Only fits ~1.5 skills
+
+ filtered_skills, was_truncated = apply_budget(skills, budget)
+ assert len(filtered_skills) == 1 # Only first skill fits
+ assert was_truncated is True
+ assert filtered_skills[0].name == "skill1"
+
+
+def test_apply_budget_alpha_strategy():
+ """Test alphabetical sorting strategy."""
+ skills = [
+ create_test_skill("zebra", "Description"),
+ create_test_skill("apple", "Description"),
+ create_test_skill("banana", "Description"),
+ ]
+ budget = SkillPromptBudget(max_chars=10000, max_skills=10, strategy="alpha")
+
+ filtered_skills, was_truncated = apply_budget(skills, budget)
+ assert len(filtered_skills) == 3
+ assert was_truncated is False
+ # Should be alphabetically sorted
+ assert filtered_skills[0].name == "apple"
+ assert filtered_skills[1].name == "banana"
+ assert filtered_skills[2].name == "zebra"
+
+
+def test_apply_budget_fifo_strategy():
+ """Test FIFO (first-in-first-out) strategy."""
+ skills = [
+ create_test_skill("first", "Description"),
+ create_test_skill("second", "Description"),
+ create_test_skill("third", "Description"),
+ ]
+ budget = SkillPromptBudget(max_chars=10000, max_skills=10, strategy="fifo")
+
+ filtered_skills, was_truncated = apply_budget(skills, budget)
+ assert len(filtered_skills) == 3
+ assert was_truncated is False
+ # Should maintain original order
+ assert filtered_skills[0].name == "first"
+ assert filtered_skills[1].name == "second"
+ assert filtered_skills[2].name == "third"
+
+
+def test_apply_budget_both_limits():
+ """Test budget when both skill count and char limits are hit."""
+ skills = [
+ create_test_skill(f"skill{i}", "A" * 50)
+ for i in range(1, 11) # 10 skills
+ ]
+ budget = SkillPromptBudget(max_chars=200, max_skills=3)
+
+ filtered_skills, was_truncated = apply_budget(skills, budget)
+ # Should be limited by character count (fewer than max_skills)
+ assert len(filtered_skills) <= 3
+ assert was_truncated is True
\ No newline at end of file
diff --git a/src/praisonai-agents/tests/unit/skills/test_events.py b/src/praisonai-agents/tests/unit/skills/test_events.py
new file mode 100644
index 000000000..0b3ca2507
--- /dev/null
+++ b/src/praisonai-agents/tests/unit/skills/test_events.py
@@ -0,0 +1,75 @@
+"""
+Tests for G-C: Skills observability events.
+"""
+
+from praisonaiagents.skills.events import SkillDiscoveredEvent, SkillActivatedEvent
+
+
+def test_skill_discovered_event():
+ """Test SkillDiscoveredEvent structure and fields."""
+ event = SkillDiscoveredEvent(
+ agent="test-agent",
+ skill_name="pdf-processing",
+ source="/home/user/skills",
+ description_chars=145
+ )
+
+ assert event.agent == "test-agent"
+ assert event.skill_name == "pdf-processing"
+ assert event.source == "/home/user/skills"
+ assert event.description_chars == 145
+
+
+def test_skill_activated_event():
+ """Test SkillActivatedEvent structure and fields."""
+ event = SkillActivatedEvent(
+ agent="test-agent",
+ skill_name="pdf-processing",
+ trigger="slash",
+ arguments="input.pdf output.txt",
+ rendered_chars=1250,
+ session_id="sess_123",
+ activation_time_ms=45.2
+ )
+
+ assert event.agent == "test-agent"
+ assert event.skill_name == "pdf-processing"
+ assert event.trigger == "slash"
+ assert event.arguments == "input.pdf output.txt"
+ assert event.rendered_chars == 1250
+ assert event.session_id == "sess_123"
+ assert event.activation_time_ms == 45.2
+
+
+def test_skill_activated_event_optional_fields():
+ """Test SkillActivatedEvent with optional fields omitted."""
+ event = SkillActivatedEvent(
+ agent="test-agent",
+ skill_name="web-scraper",
+ trigger="activate_tool",
+ arguments="https://example.com",
+ rendered_chars=890
+ )
+
+ assert event.agent == "test-agent"
+ assert event.skill_name == "web-scraper"
+ assert event.trigger == "activate_tool"
+ assert event.arguments == "https://example.com"
+ assert event.rendered_chars == 890
+ assert event.session_id is None
+ assert event.activation_time_ms is None
+
+
+def test_skill_activated_event_trigger_types():
+ """Test all valid trigger types for SkillActivatedEvent."""
+ valid_triggers = ["slash", "activate_tool", "auto"]
+
+ for trigger in valid_triggers:
+ event = SkillActivatedEvent(
+ agent="agent",
+ skill_name="skill",
+ trigger=trigger,
+ arguments="",
+ rendered_chars=100
+ )
+ assert event.trigger == trigger
\ No newline at end of file
diff --git a/src/praisonai/praisonai/capabilities/skills.py b/src/praisonai/praisonai/capabilities/skills.py
index 96c827f4b..07f2dcf1a 100644
--- a/src/praisonai/praisonai/capabilities/skills.py
+++ b/src/praisonai/praisonai/capabilities/skills.py
@@ -132,3 +132,64 @@ async def askill_load(
metadata=metadata,
**kwargs
)
+
+
+def tool_from_skill(path: str):
+ """Convert a skill to a tool function for direct tool registry integration.
+
+ This adapter allows users who prefer "skill = tool" ergonomics to
+ drop a SKILL.md into an existing tool registry without refactoring.
+
+ Args:
+ path: Path to the skill directory containing SKILL.md
+
+ Returns:
+ A function decorated with @tool that renders the skill body
+
+ Example:
+ # Convert a skill to a tool
+ pdf_tool = tool_from_skill("./skills/pdf-processing")
+
+ # Add to existing tool registry
+ agent = Agent(tools=[pdf_tool, other_tools])
+ """
+ from pathlib import Path
+ try:
+ from praisonaiagents.skills import load_skill, render_skill_body
+ from praisonaiagents.tools import tool
+
+ # Load the skill
+ skill_name = Path(path).name
+ skill_dirs = [str(Path(path).parent)]
+ loaded = load_skill(skill_name, skill_dirs)
+
+ if loaded is None:
+ raise ValueError(f"Skill not found at path: {path}")
+
+ # Create the tool function with proper metadata and argument handling
+ safe_name = skill_name.replace('-', '_').replace(' ', '_')
+ description = loaded.properties.description or f"Execute {skill_name} skill"
+
+ def _skill_tool(arguments: str = "") -> str:
+ """Execute skill with provided arguments."""
+ if loaded.instructions is None:
+ return f"Skill '{skill_name}' has no instructions"
+
+ # Return the skill instructions with argument substitution
+ return render_skill_body(loaded.instructions, arguments)
+
+ # Set dunder metadata BEFORE decoration so @tool / schema generators
+ # and test consumers both see the final identity.
+ _skill_tool.__name__ = f"skill_{safe_name}"
+ _skill_tool.__qualname__ = _skill_tool.__name__
+ _skill_tool.__doc__ = description
+
+ return tool(
+ name=f"skill_{safe_name}",
+ description=description,
+ )(_skill_tool)
+
+ except ImportError:
+ def _dummy_tool(arguments: str = "") -> str:
+ return "Skills not available (praisonaiagents not installed)"
+ return _dummy_tool
diff --git a/src/praisonai/praisonai/gateway/server.py b/src/praisonai/praisonai/gateway/server.py
index 3836d7327..1daa65ed0 100644
--- a/src/praisonai/praisonai/gateway/server.py
+++ b/src/praisonai/praisonai/gateway/server.py
@@ -1806,87 +1806,3 @@ def _request_shutdown():
if self._scheduler_task:
self._scheduler_task.cancel()
await self.stop_channels()
- except (OSError, ValueError):
- pass
-
- try:
- await _run_all()
- finally:
- if self._config_watch_task:
- self._config_watch_task.cancel()
- if self._scheduler_task:
- self._scheduler_task.cancel()
- await self.stop_channels()
- loop = asyncio.get_event_loop()
- for sig in (signal.SIGINT, signal.SIGTERM):
- try:
- loop.add_signal_handler(sig, _request_shutdown)
- except (NotImplementedError, OSError, ValueError):
- # Fallback for platforms where add_signal_handler is unavailable
- try:
- signal.signal(sig, lambda s, f: _request_shutdown())
- except (OSError, ValueError):
- pass
-
- try:
- await _run_all()
- finally:
- if self._config_watch_task:
- self._config_watch_task.cancel()
- if self._scheduler_task:
- self._scheduler_task.cancel()
- await self.stop_channels()
-
- try:
- await _run_all()
- finally:
- if self._config_watch_task:
- self._config_watch_task.cancel()
- if self._scheduler_task:
- self._scheduler_task.cancel()
- await self.stop_channels()
- loop.add_signal_handler(sig, _request_shutdown)
- except (NotImplementedError, OSError, ValueError):
- # Fallback for platforms where add_signal_handler is unavailable
- try:
- signal.signal(sig, lambda s, f: _request_shutdown())
- except (OSError, ValueError):
- pass
-
- try:
- await _run_all()
- finally:
- if self._config_watch_task:
- self._config_watch_task.cancel()
- if self._scheduler_task:
- self._scheduler_task.cancel()
- await self.stop_channels()
- self._config_watch_task.cancel()
- if self._scheduler_task:
- self._scheduler_task.cancel()
- await self.stop_channels()
-
- def _request_shutdown():
- logger.info("Received shutdown signal, stopping gateway...")
- if self._server:
- self._server.should_exit = True
-
- loop = asyncio.get_event_loop()
- for sig in (signal.SIGINT, signal.SIGTERM):
- try:
- loop.add_signal_handler(sig, _request_shutdown)
- except (NotImplementedError, OSError, ValueError):
- # Fallback for platforms where add_signal_handler is unavailable
- try:
- signal.signal(sig, lambda s, f: _request_shutdown())
- except (OSError, ValueError):
- pass
-
- try:
- await _run_all()
- finally:
- if self._config_watch_task:
- self._config_watch_task.cancel()
- if self._scheduler_task:
- self._scheduler_task.cancel()
- await self.stop_channels()
diff --git a/src/praisonai/tests/unit/test_tool_from_skill.py b/src/praisonai/tests/unit/test_tool_from_skill.py
new file mode 100644
index 000000000..0103ffafa
--- /dev/null
+++ b/src/praisonai/tests/unit/test_tool_from_skill.py
@@ -0,0 +1,115 @@
+"""
+Tests for G-F: tool_from_skill adapter.
+"""
+
+import pytest
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+from praisonai.capabilities.skills import tool_from_skill
+
+
+def test_tool_from_skill():
+ """Test tool_from_skill adapter creates a working tool function."""
+ # Create temporary skill
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ skill_dir = Path(tmp_dir) / "test-skill"
+ skill_dir.mkdir()
+
+ skill_md = skill_dir / "SKILL.md"
+ skill_md.write_text("""---
+name: test-skill
+description: A test skill for validation
+---
+
+# Test Skill
+
+This skill does testing things.
+Use it like: /test-skill arg1 arg2
+""")
+
+ # Create the tool
+ tool_func = tool_from_skill(str(skill_dir))
+
+ # Test tool properties
+ assert tool_func.__name__ == "skill_test_skill" # hyphens converted to underscores
+ assert "test skill" in tool_func.__doc__.lower()
+
+ # Test tool execution
+ result = tool_func()
+ assert "This skill does testing things" in result
+ assert "/test-skill arg1 arg2" in result
+
+
+def test_tool_from_skill_with_arguments():
+ """Test tool_from_skill with arguments parameter."""
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ skill_dir = Path(tmp_dir) / "args-skill"
+ skill_dir.mkdir()
+
+ skill_md = skill_dir / "SKILL.md"
+ skill_md.write_text("""---
+name: args-skill
+description: Skill that uses arguments
+---
+
+Process data with arguments: $ARGUMENTS
+""")
+
+ tool_func = tool_from_skill(str(skill_dir))
+
+ # Test with arguments
+ result = tool_func("file1.txt file2.txt")
+ assert "Process data with arguments" in result
+
+
+def test_tool_from_skill_nonexistent_path():
+ """Test tool_from_skill with nonexistent path raises error."""
+ with pytest.raises(ValueError, match="Skill not found at path"):
+ tool_from_skill("/nonexistent/path")
+
+
+def test_tool_from_skill_no_instructions(monkeypatch):
+ """When the loaded skill has a blank body, the tool returns a clear message."""
+ with tempfile.TemporaryDirectory() as tmp_dir:
+ skill_dir = Path(tmp_dir) / "empty-skill"
+ skill_dir.mkdir()
+ (skill_dir / "SKILL.md").write_text(
+ "---\nname: empty-skill\ndescription: Skill with no body\n---\n"
+ )
+
+ # Force instructions=None by patching the real loader.
+ import praisonaiagents.skills as skills_mod
+
+ original_load_skill = skills_mod.load_skill
+
+ def _load_with_no_instructions(name, dirs=None):
+ loaded = original_load_skill(name, dirs)
+ if loaded is not None:
+ loaded.instructions = None
+ return loaded
+
+ monkeypatch.setattr(skills_mod, "load_skill", _load_with_no_instructions)
+
+ tool_func = tool_from_skill(str(skill_dir))
+ result = tool_func()
+ assert "has no instructions" in result
+
+
+def test_tool_from_skill_import_fallback(monkeypatch):
+ """If praisonaiagents.skills is unavailable, a safe dummy tool is returned."""
+ import builtins
+
+ real_import = builtins.__import__
+
+ def _blocking_import(name, *args, **kwargs):
+ if name == "praisonaiagents.skills":
+ raise ImportError("simulated missing skills module")
+ return real_import(name, *args, **kwargs)
+
+ monkeypatch.setattr(builtins, "__import__", _blocking_import)
+
+ tool_func = tool_from_skill("/any/path")
+ result = tool_func()
+ assert "Skills not available" in result
\ No newline at end of file