heaven_base.hooks.default_hooks

 1"""Default hooks for Heaven agents — skill injection via hook system."""
 2from ..baseheavenagent import HookPoint, HookContext, HookRegistry
 3
 4
 5def make_skill_description_hook(skillset_name: str, agent_id: str = None):
 6    """BEFORE_SYSTEM_PROMPT hook: injects agent-scoped skill descriptions into system prompt.
 7
 8    Uses SkillManager with agent_id for per-agent equipped state.
 9    Equips the persona's skillset on first call, then reads from agent-scoped equipped list.
10    """
11    _initialized = [False]
12
13    def _inject_skill_descriptions(ctx: HookContext):
14        try:
15            import re
16            from skill_manager.core import SkillManager
17            manager = SkillManager(agent_id=agent_id)
18
19            # First call: equip the persona's skillset into agent-scoped state
20            if not _initialized[0] and skillset_name:
21                manager.equip_skillset(skillset_name)
22                _initialized[0] = True
23
24            equipped = manager.list_equipped()
25            if not equipped:
26                return
27
28            skill_block = "\n\n<EQUIPPED_SKILLS>\n"
29            for s in equipped:
30                desc = s.get("description", "")
31                if desc:
32                    skill_block += f"- {s['name']}: {desc}\n"
33            skill_block += "</EQUIPPED_SKILLS>"
34
35            # Strip any previous injection, then append
36            current = ctx.data.get("system_prompt", ctx.prompt or "")
37            current = re.sub(r'\n*<EQUIPPED_SKILLS>.*?</EQUIPPED_SKILLS>', '', current, flags=re.DOTALL)
38            ctx.data["system_prompt"] = current + skill_block
39        except Exception:
40            pass  # Skill injection is best-effort, never block agent
41    return _inject_skill_descriptions
42
43
44def make_skill_identity_hook(agent_name: str):
45    """BEFORE_TOOL_CALL hook: injects agent identity into sancrev treeshell skill calls.
46
47    When agent calls sancrev treeshell with skill-related actions (equip, list_equipped, etc.),
48    this hook injects the agent_name so skillmanager tracks per-agent state.
49    """
50    SKILL_ACTIONS = {"equip", "unequip", "list_equipped", "get_equipped_content", "unequip_all"}
51
52    def _inject_identity(ctx: HookContext):
53        tool_name = ctx.tool_name or ""
54        tool_args = ctx.tool_args or {}
55        # Only intercept sancrev treeshell calls
56        if "treeshell" not in tool_name.lower() and "sancrev" not in tool_name.lower():
57            return
58        command = tool_args.get("command", "")
59        # Check if this is a skill-related action
60        for action in SKILL_ACTIONS:
61            if action in command:
62                # Inject agent identity if not already present
63                if "agent_id" not in command:
64                    ctx.data["injected_identity"] = agent_name
65                break
66    return _inject_identity
67
68
69def register_skill_hooks(registry: HookRegistry, agent_name: str, skillset_name: str = None):
70    """Register both default skill hooks on a HookRegistry.
71
72    Args:
73        registry: The HookRegistry to register hooks on
74        agent_name: Agent name for identity injection and agent-scoped skill state
75        skillset_name: Optional skillset name for description injection
76    """
77    if skillset_name:
78        registry.register(HookPoint.BEFORE_SYSTEM_PROMPT, make_skill_description_hook(skillset_name, agent_id=agent_name))
79    registry.register(HookPoint.BEFORE_TOOL_CALL, make_skill_identity_hook(agent_name))