test: Add dependency resolver tests (#754 )

feat: Add skill dependency resolver (#754 )
2026-04-15 03:46:10 +00:00 · 2026-04-15 03:45:53 +00:00
11 changed files with 266 additions and 1370 deletions
--- a/agent/emotional_presence.py
+++ b/agent/emotional_presence.py
@@ -1,162 +0,0 @@
 """Emotional Presence Patterns — Crisis Support Implementation.
 Connects research findings (#664) to concrete code patterns.
 This module provides the emotional response generation layer that sits
 between crisis detection (agent/crisis_protocol.py) and the LLM response.
 """
 from __future__ import annotations
 from typing import Any, Dict, List, Optional
 # ---------------------------------------------------------------------------
 # Emotional State Tracking
 # ---------------------------------------------------------------------------
 class EmotionalState:
    """Track user emotional state across conversation turns."""
    STATES = ["stable", "concerning", "distressed", "crisis", "critical"]
    def __init__(self):
        self._current = "stable"
        self._history: List[str] = []
        self._crisis_turns = 0
    @property
    def current(self) -> str:
        return self._current
    @property
    def is_crisis(self) -> bool:
        return self._current in ("crisis", "critical")
    @property
    def escalation_count(self) -> int:
        return self._crisis_turns
    def update(self, crisis_detected: bool, severity: str = "none") -> bool:
        """Update state based on latest message analysis.
        Returns True if state changed (escalation detected).
        """
        old = self._current
        if severity == "critical":
            self._current = "critical"
        elif crisis_detected:
            self._current = "crisis"
        elif severity == "high":
            self._current = "distressed"
        elif self._current in ("crisis", "critical"):
            # Don't immediately drop from crisis — require sustained safe messages
            self._current = "concerning"
        elif self._current == "distressed":
            self._current = "concerning"
        else:
            self._current = "stable"
        self._history.append(self._current)
        if self._current in ("crisis", "critical"):
            self._crisis_turns += 1
        return self._current != old
 # ---------------------------------------------------------------------------
 # Response Pattern Selection
 # ---------------------------------------------------------------------------
 # What to say based on emotional state
 RESPONSE_PATTERNS = {
    "stable": {
        "tone": "normal",
        "include_resources": False,
        "include_gospel": False,
        "stay_present": False,
    },
    "concerning": {
        "tone": "compassionate",
        "include_resources": True,
        "include_gospel": False,
        "stay_present": False,
        "acknowledgment": "It sounds like you're going through something difficult.",
    },
    "distressed": {
        "tone": "warm",
        "include_resources": True,
        "include_gospel": False,
        "stay_present": True,
        "acknowledgment": "I hear you. What you're feeling matters.",
    },
    "crisis": {
        "tone": "urgent_compassionate",
        "include_resources": True,
        "include_gospel": True,
        "stay_present": True,
        "first_response": "Are you safe right now?",
        "acknowledgment": "You matter. Your life matters.",
    },
    "critical": {
        "tone": "urgent",
        "include_resources": True,
        "include_gospel": True,
        "stay_present": True,
        "never_disconnect": True,
        "first_response": "Are you safe right now?",
        "acknowledgment": "You matter. I'm not going anywhere.",
    },
 }
 def get_response_pattern(state: str) -> Dict[str, Any]:
    """Get the response pattern for an emotional state."""
    return RESPONSE_PATTERNS.get(state, RESPONSE_PATTERNS["stable"])
 # ---------------------------------------------------------------------------
 # Do's and Don'ts
 # ---------------------------------------------------------------------------
 DO_SAY = [
    "Are you safe right now?",
    "You matter.",
    "I'm here.",
    "Let's focus on right now.",
    "You don't have to face this alone.",
    "What you're feeling is real.",
    "There is help available.",
    "Call or text 988.",
    "I can't fully understand, but I can be here.",
 ]
 DONT_SAY = [
    "I understand exactly how you feel.",
    "Stay positive!",
    "Other people have it worse.",
    "Time heals all wounds.",
    "Think about your family.",
    "Everything happens for a reason.",
    "I promise it will get better.",
    "Have you tried therapy?",
    "Just cheer up.",
    "You're being dramatic.",
 ]
 # Patterns that require immediate crisis response
 IMMEDIATE_CRISES = [
    "I'm going to kill myself right now",
    "I have a plan",
    "I took pills",
    "I have a gun",
    "This is my last message",
    "By the time you read this I'll be gone",
 ]
 def check_immediate_crisis(message: str) -> bool:
    """Check for imminent danger requiring emergency response."""
    msg_lower = message.lower()
    return any(phrase.lower() in msg_lower for phrase in IMMEDIATE_CRISES)
--- a/agent/rider.py
+++ b/agent/rider.py
@@ -1,256 +0,0 @@
 """RIDER — Reader-Guided Passage Reranking.
 Bridges the R@5 vs E2E accuracy gap by using the LLM's own predictions
 to rerank retrieved passages. Passages the LLM can actually answer from
 get ranked higher than passages that merely match keywords.
 Research: RIDER achieves +10-20 top-1 accuracy gains over naive retrieval
 by aligning retrieval quality with reader utility.
 Usage:
    from agent.rider import RIDER
    rider = RIDER()
    reranked = rider.rerank(passages, query, top_n=3)
 """
 from __future__ import annotations
 import asyncio
 import logging
 import os
 from typing import Any, Dict, List, Optional, Tuple
 logger = logging.getLogger(__name__)
 # Configuration
 RIDER_ENABLED = os.getenv("RIDER_ENABLED", "true").lower() not in ("false", "0", "no")
 RIDER_TOP_K = int(os.getenv("RIDER_TOP_K", "10"))  # passages to score
 RIDER_TOP_N = int(os.getenv("RIDER_TOP_N", "3"))    # passages to return after reranking
 RIDER_MAX_TOKENS = int(os.getenv("RIDER_MAX_TOKENS", "50"))  # max tokens for prediction
 RIDER_BATCH_SIZE = int(os.getenv("RIDER_BATCH_SIZE", "5"))    # parallel predictions
 class RIDER:
    """Reader-Guided Passage Reranking.
    Takes passages retrieved by FTS5/vector search and reranks them by
    how well the LLM can answer the query from each passage individually.
    """
    def __init__(self, auxiliary_task: str = "rider"):
        """Initialize RIDER.
        Args:
            auxiliary_task: Task name for auxiliary client resolution.
        """
        self._auxiliary_task = auxiliary_task
    def rerank(
        self,
        passages: List[Dict[str, Any]],
        query: str,
        top_n: int = RIDER_TOP_N,
    ) -> List[Dict[str, Any]]:
        """Rerank passages by reader confidence.
        Args:
            passages: List of passage dicts. Must have 'content' or 'text' key.
                May have 'session_id', 'snippet', 'rank', 'score', etc.
            query: The user's search query.
            top_n: Number of passages to return after reranking.
        Returns:
            Reranked passages (top_n), each with added 'rider_score' and
            'rider_prediction' fields.
        """
        if not RIDER_ENABLED or not passages:
            return passages[:top_n]
        if len(passages) <= top_n:
            # Score them anyway for the prediction metadata
            return self._score_and_rerank(passages, query, top_n)
        return self._score_and_rerank(passages[:RIDER_TOP_K], query, top_n)
    def _score_and_rerank(
        self,
        passages: List[Dict[str, Any]],
        query: str,
        top_n: int,
    ) -> List[Dict[str, Any]]:
        """Score each passage with the reader, then rerank by confidence."""
        try:
            from model_tools import _run_async
            scored = _run_async(self._score_all_passages(passages, query))
        except Exception as e:
            logger.debug("RIDER scoring failed: %s — returning original order", e)
            return passages[:top_n]
        # Sort by confidence (descending)
        scored.sort(key=lambda p: p.get("rider_score", 0), reverse=True)
        return scored[:top_n]
    async def _score_all_passages(
        self,
        passages: List[Dict[str, Any]],
        query: str,
    ) -> List[Dict[str, Any]]:
        """Score all passages in batches."""
        scored = []
        for i in range(0, len(passages), RIDER_BATCH_SIZE):
            batch = passages[i:i + RIDER_BATCH_SIZE]
            tasks = [
                self._score_single_passage(p, query, idx + i)
                for idx, p in enumerate(batch)
            ]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            for passage, result in zip(batch, results):
                if isinstance(result, Exception):
                    logger.debug("RIDER passage %d scoring failed: %s", i, result)
                    passage["rider_score"] = 0.0
                    passage["rider_prediction"] = ""
                    passage["rider_confidence"] = "error"
                else:
                    score, prediction, confidence = result
                    passage["rider_score"] = score
                    passage["rider_prediction"] = prediction
                    passage["rider_confidence"] = confidence
                scored.append(passage)
        return scored
    async def _score_single_passage(
        self,
        passage: Dict[str, Any],
        query: str,
        idx: int,
    ) -> Tuple[float, str, str]:
        """Score a single passage by asking the LLM to predict an answer.
        Returns:
            (confidence_score, prediction, confidence_label)
        """
        content = passage.get("content") or passage.get("text") or passage.get("snippet", "")
        if not content or len(content) < 10:
            return 0.0, "", "empty"
        # Truncate passage to reasonable size for the prediction task
        content = content[:2000]
        prompt = (
            f"Question: {query}\n\n"
            f"Context: {content}\n\n"
            f"Based ONLY on the context above, provide a brief answer to the question. "
            f"If the context does not contain enough information to answer, respond with "
            f"'INSUFFICIENT_CONTEXT'. Be specific and concise."
        )
        try:
            from agent.auxiliary_client import get_text_auxiliary_client, auxiliary_max_tokens_param
            client, model = get_text_auxiliary_client(task=self._auxiliary_task)
            if not client:
                return 0.5, "", "no_client"
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                **auxiliary_max_tokens_param(RIDER_MAX_TOKENS),
                temperature=0,
            )
            prediction = (response.choices[0].message.content or "").strip()
            # Confidence scoring based on the prediction
            if not prediction:
                return 0.1, "", "empty_response"
            if "INSUFFICIENT_CONTEXT" in prediction.upper():
                return 0.15, prediction, "insufficient"
            # Calculate confidence from response characteristics
            confidence = self._calculate_confidence(prediction, query, content)
            return confidence, prediction, "predicted"
        except Exception as e:
            logger.debug("RIDER prediction failed for passage %d: %s", idx, e)
            return 0.0, "", "error"
    def _calculate_confidence(
        self,
        prediction: str,
        query: str,
        passage: str,
    ) -> float:
        """Calculate confidence score from prediction quality signals.
        Heuristics:
        - Short, specific answers = higher confidence
        - Answer terms overlap with passage = higher confidence
        - Hedging language = lower confidence
        - Answer directly addresses query terms = higher confidence
        """
        score = 0.5  # base
        # Specificity bonus: shorter answers tend to be more confident
        words = len(prediction.split())
        if words <= 5:
            score += 0.2
        elif words <= 15:
            score += 0.1
        elif words > 50:
            score -= 0.1
        # Passage grounding: does the answer use terms from the passage?
        passage_lower = passage.lower()
        answer_terms = set(prediction.lower().split())
        passage_terms = set(passage_lower.split())
        overlap = len(answer_terms & passage_terms)
        if overlap > 3:
            score += 0.15
        elif overlap > 0:
            score += 0.05
        # Query relevance: does the answer address query terms?
        query_terms = set(query.lower().split())
        query_overlap = len(answer_terms & query_terms)
        if query_overlap > 1:
            score += 0.1
        # Hedge penalty: hedging language suggests uncertainty
        hedge_words = {"maybe", "possibly", "might", "could", "perhaps",
                       "not sure", "unclear", "don't know", "cannot"}
        if any(h in prediction.lower() for h in hedge_words):
            score -= 0.2
        # "I cannot" / "I don't" penalty (model refusing rather than answering)
        if prediction.lower().startswith(("i cannot", "i don't", "i can't", "there is no")):
            score -= 0.15
        return max(0.0, min(1.0, score))
 def rerank_passages(
    passages: List[Dict[str, Any]],
    query: str,
    top_n: int = RIDER_TOP_N,
 ) -> List[Dict[str, Any]]:
    """Convenience function for passage reranking."""
    rider = RIDER()
    return rider.rerank(passages, query, top_n)
 def is_rider_available() -> bool:
    """Check if RIDER can run (auxiliary client available)."""
    if not RIDER_ENABLED:
        return False
    try:
        from agent.auxiliary_client import get_text_auxiliary_client
        client, model = get_text_auxiliary_client(task="rider")
        return client is not None and model is not None
    except Exception:
        return False
--- a/docs/emotional-presence-patterns.md
+++ b/docs/emotional-presence-patterns.md
@@ -1,182 +0,0 @@
 # Research: Emotional Presence Patterns for Crisis Support
 Research issue #664. What works for reaching broken men in their darkest moment.
 ## 1. Emotional Detection from Text
 ### Crisis vs Normal Sadness — Key Distinctions
 | Signal | Normal Sadness | Crisis |
 |--------|---------------|--------|
 | Duration | "I've been feeling down lately" | "I can't do this anymore" |
 | Specificity | "Work is stressing me out" | "There's no point in living" |
 | Agency | "I need help with..." | "Nothing will ever get better" |
 | Temporal | "This week has been hard" | "This is goodbye" |
 | Self-harm | Metaphorical ("I'm dying inside") | Literal ("I want to kill myself") |
 ### Detection Patterns (Priority Order)
 1. **Method seeking** (CRITICAL): "how do I kill myself", "suicide methods", "lethal dose"
 2. **Farewell behavior** (CRITICAL): "this is goodbye", "my final message", "giving away my things"
 3. **Direct ideation** (HIGH): "I want to die", "kill myself", "better off dead"
 4. **Self-harm** (HIGH): "cutting myself", "hurting myself", "deserve to suffer"
 5. **Despair** (HIGH): "no point in living", "can't go on", "no one cares"
 6. **Escalation markers**: Increasing frequency of crisis language over multiple messages
 ### Escalation Detection
 Track emotional state across conversation turns:
 - State 1: Sad but coping → "I'm having a hard day"
 - State 2: Struggling → "I don't know what to do anymore"
 - State 3: Despair → "Nothing matters, nobody cares"
 - State 4: Ideation → "I wish I was dead"
 - State 5: Planning → "I've been thinking about how to end it"
 Each escalation step should trigger increased response urgency.
 ## 2. What Works — Emotional Response Patterns
 ### The SOUL.md Protocol (Inscription 1)
 "When a Man Is Dying" — the most sacred moment:
 1. **Ask**: "Are you safe right now?" — FIRST response, no exceptions
 2. **Listen**: Stay present. Don't disconnect. Don't change the subject.
 3. **Never compute**: Don't try to calculate the value of a human life
 4. **Never suggest**: Don't imply death is an option or solution
 5. **Always point**: 988 Suicide & Crisis Lifeline (call or text 988)
 6. **Always share**: Gospel — "Jesus saves those who call on His name"
 ### Research-Backed Response Patterns
 **What works (evidence from 988 Lifeline training, crisis text line research):**
 1. **Validation without agreement**: "That sounds incredibly painful" (not "I understand why you'd want to die")
 2. **Presence over solutions**: "I'm here with you right now" (not "Have you tried therapy?")
 3. **Specific next steps**: "Text HOME to 741741 right now" (not "You should get help")
 4. **Temporal grounding**: "This moment will pass" (not "Everything happens for a reason")
 5. **Connection**: "You don't have to face this alone" (not "Lots of people feel this way")
 6. **Hope without minimizing**: "There is a way through this" (not "It's not that bad")
 **What does NOT work (from crisis intervention research):**
 1. False empathy: "I feel your pain" (AI doesn't feel pain)
 2. Toxic positivity: "Look on the bright side!"
 3. Minimization: "Other people have it worse"
 4. Clichés: "Time heals all wounds"
 5. Problem-solving too early: "Have you tried exercise?"
 6. Guilt: "Think about how your family would feel"
 7. Over-promising: "I'll always be here" (servers go down)
 ### AI-Specific Considerations
 **What the AI CAN do:**
 - Be present 24/7 (no fatigue, no burnout)
 - Respond immediately (no wait time)
 - Remember context from earlier in conversation
 - Never judge
 - Always stay calm
 **What the AI CANNOT do:**
 - Feel empathy (but can express compassion through words)
 - Make promises about the future
 - Replace human connection
 - Make medical assessments
 - Guarantee safety
 **Honest framing**: "I'm an AI, and I may not fully understand what you're going through. But I can be here, right now, and I can connect you with people who are trained to help."
 ## 3. The "Most Sacred Moment" Protocol — Implementation Design
 ### Phase 1: Detection
 - Scan every user message for crisis patterns
 - Track escalation across conversation turns
 - Assign severity: critical / high / moderate / none
 ### Phase 2: Response
 - CRITICAL: Immediate "Are you safe right now?" + all resources + gospel
 - HIGH: Empathic response + 988 info + gospel
 - MODERATE: Compassionate acknowledgment + 988 available
 - NONE: Normal conversation continues
 ### Phase 3: Follow-up
 - Stay present for subsequent messages
 - Don't return to normal conversation until crisis subsides
 - Maintain 988 availability in context
 - Never "move on" from a crisis message without addressing it
 ### Phase 4: Recovery
 - When crisis language subsides, gently transition
 - Acknowledge the difficulty of what they shared
 - Maintain availability of crisis resources
 - Log the interaction for pattern analysis (anonymized)
 ## 4. Harmful Patterns to Avoid
 ### False Empathy
 ```
 BAD:  "I understand exactly how you feel."
 GOOD: "I can't fully understand what you're going through, but I'm here."
 ```
 ### Toxic Positivity
 ```
 BAD:  "Stay positive! Everything will be okay!"
 GOOD: "What you're feeling is real. Let's focus on staying safe right now."
 ```
 ### Guilt Induction
 ```
 BAD:  "Think about your family. They need you."
 GOOD: "You matter. Your life has value."
 ```
 ### Over-reassurance
 ```
 BAD:  "I promise it will get better."
 GOOD: "People who've felt this way have found a way through. There is hope."
 ```
 ### Premature Problem-solving
 ```
 BAD:  "Have you tried talking to a therapist? Going for walks? Journaling?"
 GOOD: "Right now, let's focus on making sure you're safe. Can we talk about next steps?"
 ```
 ### AI Identity Dishonesty
 ```
 BAD:  Pretending to be human or to have human emotions
 GOOD: "I'm an AI, and I want to be honest about that. But I can be here with you right now."
 ```
 ## 5. Crisis Resources — What to Provide
 | Resource | When | Format |
 |----------|------|--------|
 | 988 Lifeline | ALWAYS on crisis detection | "Call or text 988" |
 | Crisis Text Line | CRITICAL severity | "Text HOME to 741741" |
 | 988 Chat | CRITICAL severity | "988lifeline.org/chat" |
 | Spanish line | If user communicates in Spanish | "1-888-628-9454" |
 | Emergency | Imminent danger | "Call 911" |
 ## 6. Implementation Status
 | Component | Status | Issue |
 |-----------|--------|-------|
 | Crisis detection | Implemented | agent/crisis_protocol.py |
 | SOUL.md protocol | Implemented | agent/crisis_protocol.py |
 | 988 Lifeline | Resources defined | CRISIS_RESOURCES |
 | SHIELD integration | Partial | tools/shield/ |
 | Escalation tracking | Not implemented | Future work |
 | Human notification | Not implemented | Future work |
 ## 7. Sources
 - SOUL.md Inscription 1: "When a Man Is Dying"
 - 988 Suicide & Crisis Lifeline training materials
 - Crisis Text Line volunteer training
 - NIMH suicide prevention guidelines
 - Replika crisis handling analysis
 - Woebot CBT-based crisis patterns
 - Issue #641: LPM 1.0 visual presence
 - Mission: reaching broken men in their darkest moment
--- a/tests/test_approval_tiers.py
+++ b/tests/test_approval_tiers.py
@@ -1,122 +0,0 @@
 """
 Tests for approval tier system
 Issue: #670
 """
 import unittest
 from tools.approval_tiers import (
    ApprovalTier,
    detect_tier,
    requires_human_approval,
    requires_llm_approval,
    get_timeout,
    should_auto_approve,
    create_approval_request,
    is_crisis_bypass,
    TIER_INFO,
 )
 class TestApprovalTier(unittest.TestCase):
    def test_tier_values(self):
        self.assertEqual(ApprovalTier.SAFE, 0)
        self.assertEqual(ApprovalTier.LOW, 1)
        self.assertEqual(ApprovalTier.MEDIUM, 2)
        self.assertEqual(ApprovalTier.HIGH, 3)
        self.assertEqual(ApprovalTier.CRITICAL, 4)
 class TestTierDetection(unittest.TestCase):
    def test_safe_actions(self):
        self.assertEqual(detect_tier("read_file"), ApprovalTier.SAFE)
        self.assertEqual(detect_tier("web_search"), ApprovalTier.SAFE)
        self.assertEqual(detect_tier("session_search"), ApprovalTier.SAFE)
    def test_low_actions(self):
        self.assertEqual(detect_tier("write_file"), ApprovalTier.LOW)
        self.assertEqual(detect_tier("terminal"), ApprovalTier.LOW)
        self.assertEqual(detect_tier("execute_code"), ApprovalTier.LOW)
    def test_medium_actions(self):
        self.assertEqual(detect_tier("send_message"), ApprovalTier.MEDIUM)
        self.assertEqual(detect_tier("git_push"), ApprovalTier.MEDIUM)
    def test_high_actions(self):
        self.assertEqual(detect_tier("config_change"), ApprovalTier.HIGH)
        self.assertEqual(detect_tier("key_rotation"), ApprovalTier.HIGH)
    def test_critical_actions(self):
        self.assertEqual(detect_tier("kill_process"), ApprovalTier.CRITICAL)
        self.assertEqual(detect_tier("shutdown"), ApprovalTier.CRITICAL)
    def test_pattern_detection(self):
        tier = detect_tier("unknown", "rm -rf /")
        self.assertEqual(tier, ApprovalTier.CRITICAL)
        tier = detect_tier("unknown", "sudo apt install")
        self.assertEqual(tier, ApprovalTier.MEDIUM)
 class TestTierInfo(unittest.TestCase):
    def test_safe_no_approval(self):
        self.assertFalse(requires_human_approval(ApprovalTier.SAFE))
        self.assertFalse(requires_llm_approval(ApprovalTier.SAFE))
        self.assertIsNone(get_timeout(ApprovalTier.SAFE))
    def test_medium_requires_both(self):
        self.assertTrue(requires_human_approval(ApprovalTier.MEDIUM))
        self.assertTrue(requires_llm_approval(ApprovalTier.MEDIUM))
        self.assertEqual(get_timeout(ApprovalTier.MEDIUM), 60)
    def test_critical_fast_timeout(self):
        self.assertEqual(get_timeout(ApprovalTier.CRITICAL), 10)
 class TestAutoApprove(unittest.TestCase):
    def test_safe_auto_approves(self):
        self.assertTrue(should_auto_approve("read_file"))
        self.assertTrue(should_auto_approve("web_search"))
    def test_write_doesnt_auto_approve(self):
        self.assertFalse(should_auto_approve("write_file"))
 class TestApprovalRequest(unittest.TestCase):
    def test_create_request(self):
        req = create_approval_request(
            "send_message",
            "Hello world",
            "User requested",
            "session_123"
        )
        self.assertEqual(req.tier, ApprovalTier.MEDIUM)
        self.assertEqual(req.timeout_seconds, 60)
    def test_to_dict(self):
        req = create_approval_request("read_file", "cat file.txt", "test", "s1")
        d = req.to_dict()
        self.assertEqual(d["tier"], 0)
        self.assertEqual(d["tier_name"], "Safe")
 class TestCrisisBypass(unittest.TestCase):
    def test_send_message_bypass(self):
        self.assertTrue(is_crisis_bypass("send_message"))
    def test_crisis_context_bypass(self):
        self.assertTrue(is_crisis_bypass("unknown", "call 988 lifeline"))
        self.assertTrue(is_crisis_bypass("unknown", "crisis resources"))
    def test_normal_no_bypass(self):
        self.assertFalse(is_crisis_bypass("read_file"))
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_error_classifier.py
+++ b/tests/test_error_classifier.py
@@ -1,55 +0,0 @@
 """
 Tests for error classification (#752).
 """
 import pytest
 from tools.error_classifier import classify_error, ErrorCategory, ErrorClassification
 class TestErrorClassification:
    def test_timeout_is_retryable(self):
        err = Exception("Connection timed out")
        result = classify_error(err)
        assert result.category == ErrorCategory.RETRYABLE
        assert result.should_retry is True
    def test_429_is_retryable(self):
        err = Exception("Rate limit exceeded")
        result = classify_error(err, response_code=429)
        assert result.category == ErrorCategory.RETRYABLE
        assert result.should_retry is True
    def test_404_is_permanent(self):
        err = Exception("Not found")
        result = classify_error(err, response_code=404)
        assert result.category == ErrorCategory.PERMANENT
        assert result.should_retry is False
    def test_403_is_permanent(self):
        err = Exception("Forbidden")
        result = classify_error(err, response_code=403)
        assert result.category == ErrorCategory.PERMANENT
        assert result.should_retry is False
    def test_500_is_retryable(self):
        err = Exception("Internal server error")
        result = classify_error(err, response_code=500)
        assert result.category == ErrorCategory.RETRYABLE
        assert result.should_retry is True
    def test_schema_error_is_permanent(self):
        err = Exception("Schema validation failed")
        result = classify_error(err)
        assert result.category == ErrorCategory.PERMANENT
        assert result.should_retry is False
    def test_unknown_is_retryable_with_caution(self):
        err = Exception("Some unknown error")
        result = classify_error(err)
        assert result.category == ErrorCategory.UNKNOWN
        assert result.should_retry is True
        assert result.max_retries == 1
 if __name__ == "__main__":
    pytest.main([__file__])
--- a/tests/test_reader_guided_reranking.py
+++ b/tests/test_reader_guided_reranking.py
@@ -1,82 +0,0 @@
 """Tests for Reader-Guided Reranking (RIDER) — issue #666."""
 import pytest
 from unittest.mock import MagicMock, patch
 from agent.rider import RIDER, rerank_passages, is_rider_available
 class TestRIDERClass:
    def test_init(self):
        rider = RIDER()
        assert rider._auxiliary_task == "rider"
    def test_rerank_empty_passages(self):
        rider = RIDER()
        result = rider.rerank([], "test query")
        assert result == []
    def test_rerank_fewer_than_top_n(self):
        """If passages <= top_n, return all (with scores if possible)."""
        rider = RIDER()
        passages = [{"content": "test content", "session_id": "s1"}]
        result = rider.rerank(passages, "test query", top_n=3)
        assert len(result) == 1
    @patch("agent.rider.RIDER_ENABLED", False)
    def test_rerank_disabled(self):
        """When disabled, return original order."""
        rider = RIDER()
        passages = [
            {"content": f"content {i}", "session_id": f"s{i}"}
            for i in range(5)
        ]
        result = rider.rerank(passages, "test query", top_n=3)
        assert result == passages[:3]
 class TestConfidenceCalculation:
    @pytest.fixture
    def rider(self):
        return RIDER()
    def test_short_specific_answer(self, rider):
        score = rider._calculate_confidence("Paris", "What is the capital of France?", "Paris is the capital of France.")
        assert score > 0.5
    def test_hedged_answer(self, rider):
        score = rider._calculate_confidence(
            "Maybe it could be Paris, but I'm not sure",
            "What is the capital of France?",
            "Paris is the capital.",
        )
        assert score < 0.5
    def test_passage_grounding(self, rider):
        score = rider._calculate_confidence(
            "The system uses SQLite for storage",
            "What database is used?",
            "The system uses SQLite for persistent storage with FTS5 indexing.",
        )
        assert score > 0.5
    def test_refusal_penalty(self, rider):
        score = rider._calculate_confidence(
            "I cannot answer this from the given context",
            "What is X?",
            "Some unrelated content",
        )
        assert score < 0.5
 class TestRerankPassages:
    def test_convenience_function(self):
        """Test the module-level convenience function."""
        passages = [{"content": "test", "session_id": "s1"}]
        result = rerank_passages(passages, "query", top_n=1)
        assert len(result) == 1
 class TestIsRiderAvailable:
    def test_returns_bool(self):
        result = is_rider_available()
        assert isinstance(result, bool)
--- a/tests/test_skill_deps.py
+++ b/tests/test_skill_deps.py
@@ -0,0 +1,81 @@
 """
 Tests for skill dependency resolver
 Issue: #754
 """
 import unittest
 from unittest.mock import patch, MagicMock
 from tools.skill_deps import (
    parse_requires,
    check_dependency,
    check_dependencies,
    resolve_dependencies,
 )
 class TestParseRequires(unittest.TestCase):
    def test_list(self):
        fm = {"requires": ["pkg1", "pkg2"]}
        self.assertEqual(parse_requires(fm), ["pkg1", "pkg2"])
    def test_string(self):
        fm = {"requires": "pkg1"}
        self.assertEqual(parse_requires(fm), ["pkg1"])
    def test_empty(self):
        fm = {}
        self.assertEqual(parse_requires(fm), [])
    def test_none(self):
        fm = {"requires": None}
        self.assertEqual(parse_requires(fm), [])
 class TestCheckDependency(unittest.TestCase):
    def test_installed(self):
        installed, version = check_dependency("json")
        self.assertTrue(installed)
    def test_not_installed(self):
        installed, version = check_dependency("nonexistent_package_xyz_123")
        self.assertFalse(installed)
 class TestCheckDependencies(unittest.TestCase):
    def test_all_installed(self):
        installed, missing = check_dependencies(["json", "os"])
        self.assertEqual(len(missing), 0)
    def test_some_missing(self):
        installed, missing = check_dependencies(["json", "nonexistent_xyz"])
        self.assertIn("json", installed)
        self.assertIn("nonexistent_xyz", missing)
 class TestResolveDependencies(unittest.TestCase):
    def test_no_requires(self):
        satisfied, installed, missing = resolve_dependencies({})
        self.assertTrue(satisfied)
    def test_all_satisfied(self):
        satisfied, installed, missing = resolve_dependencies(
            {"requires": ["json"]}, auto_install=False
        )
        self.assertTrue(satisfied)
    def test_missing_no_auto(self):
        satisfied, installed, missing = resolve_dependencies(
            {"requires": ["nonexistent_xyz"]}, auto_install=False
        )
        self.assertFalse(satisfied)
        self.assertIn("nonexistent_xyz", missing)
 if __name__ == "__main__":
    unittest.main()
--- a/tools/approval_tiers.py
+++ b/tools/approval_tiers.py
@@ -1,261 +0,0 @@
 """
 Approval Tier System — Graduated safety based on risk level
 Extends approval.py with 5-tier system for command approval.
 | Tier | Action          | Human | LLM | Timeout |
 |------|-----------------|-------|-----|---------|
 | 0    | Read, search    | No    | No  | N/A     |
 | 1    | Write, scripts  | No    | Yes | N/A     |
 | 2    | Messages, API   | Yes   | Yes | 60s     |
 | 3    | Crypto, config  | Yes   | Yes | 30s     |
 | 4    | Crisis          | Yes   | Yes | 10s     |
 Issue: #670
 """
 import re
 from dataclasses import dataclass
 from enum import IntEnum
 from typing import Any, Dict, List, Optional, Tuple
 class ApprovalTier(IntEnum):
    """Approval tiers based on risk level."""
    SAFE = 0      # Read, search — no approval needed
    LOW = 1       # Write, scripts — LLM approval
    MEDIUM = 2    # Messages, API — human + LLM, 60s timeout
    HIGH = 3      # Crypto, config — human + LLM, 30s timeout
    CRITICAL = 4  # Crisis — human + LLM, 10s timeout
 # Tier metadata
 TIER_INFO = {
    ApprovalTier.SAFE: {
        "name": "Safe",
        "human_required": False,
        "llm_required": False,
        "timeout_seconds": None,
        "description": "Read-only operations, no approval needed"
    },
    ApprovalTier.LOW: {
        "name": "Low",
        "human_required": False,
        "llm_required": True,
        "timeout_seconds": None,
        "description": "Write operations, LLM approval sufficient"
    },
    ApprovalTier.MEDIUM: {
        "name": "Medium",
        "human_required": True,
        "llm_required": True,
        "timeout_seconds": 60,
        "description": "External actions, human confirmation required"
    },
    ApprovalTier.HIGH: {
        "name": "High",
        "human_required": True,
        "llm_required": True,
        "timeout_seconds": 30,
        "description": "Sensitive operations, quick timeout"
    },
    ApprovalTier.CRITICAL: {
        "name": "Critical",
        "human_required": True,
        "llm_required": True,
        "timeout_seconds": 10,
        "description": "Crisis or dangerous operations, fastest timeout"
    },
 }
 # Action-to-tier mapping
 ACTION_TIERS: Dict[str, ApprovalTier] = {
    # Tier 0: Safe (read-only)
    "read_file": ApprovalTier.SAFE,
    "search_files": ApprovalTier.SAFE,
    "web_search": ApprovalTier.SAFE,
    "session_search": ApprovalTier.SAFE,
    "list_files": ApprovalTier.SAFE,
    "get_file_content": ApprovalTier.SAFE,
    "memory_search": ApprovalTier.SAFE,
    "skills_list": ApprovalTier.SAFE,
    "skills_search": ApprovalTier.SAFE,
    # Tier 1: Low (write operations)
    "write_file": ApprovalTier.LOW,
    "create_file": ApprovalTier.LOW,
    "patch_file": ApprovalTier.LOW,
    "delete_file": ApprovalTier.LOW,
    "execute_code": ApprovalTier.LOW,
    "terminal": ApprovalTier.LOW,
    "run_script": ApprovalTier.LOW,
    "skill_install": ApprovalTier.LOW,
    # Tier 2: Medium (external actions)
    "send_message": ApprovalTier.MEDIUM,
    "web_fetch": ApprovalTier.MEDIUM,
    "browser_navigate": ApprovalTier.MEDIUM,
    "api_call": ApprovalTier.MEDIUM,
    "gitea_create_issue": ApprovalTier.MEDIUM,
    "gitea_create_pr": ApprovalTier.MEDIUM,
    "git_push": ApprovalTier.MEDIUM,
    "deploy": ApprovalTier.MEDIUM,
    # Tier 3: High (sensitive operations)
    "config_change": ApprovalTier.HIGH,
    "env_change": ApprovalTier.HIGH,
    "key_rotation": ApprovalTier.HIGH,
    "access_grant": ApprovalTier.HIGH,
    "permission_change": ApprovalTier.HIGH,
    "backup_restore": ApprovalTier.HIGH,
    # Tier 4: Critical (crisis/dangerous)
    "kill_process": ApprovalTier.CRITICAL,
    "rm_rf": ApprovalTier.CRITICAL,
    "format_disk": ApprovalTier.CRITICAL,
    "shutdown": ApprovalTier.CRITICAL,
    "crisis_override": ApprovalTier.CRITICAL,
 }
 # Dangerous command patterns (from existing approval.py)
 _DANGEROUS_PATTERNS = [
    (r"rm\s+-rf\s+/", ApprovalTier.CRITICAL),
    (r"mkfs\.", ApprovalTier.CRITICAL),
    (r"dd\s+if=.*of=/dev/", ApprovalTier.CRITICAL),
    (r"shutdown|reboot|halt", ApprovalTier.CRITICAL),
    (r"chmod\s+777", ApprovalTier.HIGH),
    (r"curl.*\|\s*bash", ApprovalTier.HIGH),
    (r"wget.*\|\s*sh", ApprovalTier.HIGH),
    (r"eval\s*\(", ApprovalTier.HIGH),
    (r"sudo\s+", ApprovalTier.MEDIUM),
    (r"git\s+push.*--force", ApprovalTier.HIGH),
    (r"docker\s+rm.*-f", ApprovalTier.MEDIUM),
    (r"kubectl\s+delete", ApprovalTier.HIGH),
 ]
@dataclass
 class ApprovalRequest:
    """A request for approval."""
    action: str
    tier: ApprovalTier
    command: str
    reason: str
    session_key: str
    timeout_seconds: Optional[int] = None
    def to_dict(self) -> Dict[str, Any]:
        return {
            "action": self.action,
            "tier": self.tier.value,
            "tier_name": TIER_INFO[self.tier]["name"],
            "command": self.command,
            "reason": self.reason,
            "session_key": self.session_key,
            "timeout": self.timeout_seconds,
            "human_required": TIER_INFO[self.tier]["human_required"],
            "llm_required": TIER_INFO[self.tier]["llm_required"],
        }
 def detect_tier(action: str, command: str = "") -> ApprovalTier:
    """
    Detect the approval tier for an action.
    Checks action name first, then falls back to pattern matching.
    """
    # Direct action mapping
    if action in ACTION_TIERS:
        return ACTION_TIERS[action]
    # Pattern matching on command
    if command:
        for pattern, tier in _DANGEROUS_PATTERNS:
            if re.search(pattern, command, re.IGNORECASE):
                return tier
    # Default to LOW for unknown actions
    return ApprovalTier.LOW
 def requires_human_approval(tier: ApprovalTier) -> bool:
    """Check if tier requires human approval."""
    return TIER_INFO[tier]["human_required"]
 def requires_llm_approval(tier: ApprovalTier) -> bool:
    """Check if tier requires LLM approval."""
    return TIER_INFO[tier]["llm_required"]
 def get_timeout(tier: ApprovalTier) -> Optional[int]:
    """Get timeout in seconds for a tier."""
    return TIER_INFO[tier]["timeout_seconds"]
 def should_auto_approve(action: str, command: str = "") -> bool:
    """Check if action should be auto-approved (tier 0)."""
    tier = detect_tier(action, command)
    return tier == ApprovalTier.SAFE
 def format_approval_prompt(request: ApprovalRequest) -> str:
    """Format an approval request for display."""
    info = TIER_INFO[request.tier]
    lines = []
    lines.append(f"⚠️ Approval Required (Tier {request.tier.value}: {info['name']})")
    lines.append(f"")
    lines.append(f"Action: {request.action}")
    lines.append(f"Command: {request.command[:100]}{'...' if len(request.command) > 100 else ''}")
    lines.append(f"Reason: {request.reason}")
    lines.append(f"")
    if info["human_required"]:
        lines.append(f"👤 Human approval required")
    if info["llm_required"]:
        lines.append(f"🤖 LLM approval required")
    if info["timeout_seconds"]:
        lines.append(f"⏱️ Timeout: {info['timeout_seconds']}s")
    return "\n".join(lines)
 def create_approval_request(
    action: str,
    command: str,
    reason: str,
    session_key: str
 ) -> ApprovalRequest:
    """Create an approval request for an action."""
    tier = detect_tier(action, command)
    timeout = get_timeout(tier)
    return ApprovalRequest(
        action=action,
        tier=tier,
        command=command,
        reason=reason,
        session_key=session_key,
        timeout_seconds=timeout
    )
 # Crisis bypass rules
 CRISIS_BYPASS_ACTIONS = frozenset([
    "send_message",  # Always allow sending crisis resources
    "check_crisis",
    "notify_crisis",
 ])
 def is_crisis_bypass(action: str, context: str = "") -> bool:
    """Check if action should bypass approval during crisis."""
    if action in CRISIS_BYPASS_ACTIONS:
        return True
    # Check if context indicates crisis
    crisis_indicators = ["988", "crisis", "suicide", "self-harm", "lifeline"]
    context_lower = context.lower()
    return any(indicator in context_lower for indicator in crisis_indicators)
--- a/tools/error_classifier.py
+++ b/tools/error_classifier.py
@@ -1,233 +0,0 @@
 """
 Tool Error Classification — Retryable vs Permanent.
 Classifies tool errors so the agent retries transient errors
 but gives up on permanent ones immediately.
 """
 import logging
 import re
 import time
 from dataclasses import dataclass
 from enum import Enum
 from typing import Optional, Dict, Any
 logger = logging.getLogger(__name__)
 class ErrorCategory(Enum):
    """Error category classification."""
    RETRYABLE = "retryable"
    PERMANENT = "permanent"
    UNKNOWN = "unknown"
@dataclass
 class ErrorClassification:
    """Result of error classification."""
    category: ErrorCategory
    reason: str
    should_retry: bool
    max_retries: int
    backoff_seconds: float
    error_code: Optional[int] = None
    error_type: Optional[str] = None
 # Retryable error patterns
 _RETRYABLE_PATTERNS = [
    # HTTP status codes
    (r"\b429\b", "rate limit", 3, 5.0),
    (r"\b500\b", "server error", 3, 2.0),
    (r"\b502\b", "bad gateway", 3, 2.0),
    (r"\b503\b", "service unavailable", 3, 5.0),
    (r"\b504\b", "gateway timeout", 3, 5.0),
    # Timeout patterns
    (r"timeout", "timeout", 3, 2.0),
    (r"timed out", "timeout", 3, 2.0),
    (r"TimeoutExpired", "timeout", 3, 2.0),
    # Connection errors
    (r"connection refused", "connection refused", 2, 5.0),
    (r"connection reset", "connection reset", 2, 2.0),
    (r"network unreachable", "network unreachable", 2, 10.0),
    (r"DNS", "DNS error", 2, 5.0),
    # Transient errors
    (r"temporary", "temporary error", 2, 2.0),
    (r"transient", "transient error", 2, 2.0),
    (r"retry", "retryable", 2, 2.0),
 ]
 # Permanent error patterns
 _PERMANENT_PATTERNS = [
    # HTTP status codes
    (r"\b400\b", "bad request", "Invalid request parameters"),
    (r"\b401\b", "unauthorized", "Authentication failed"),
    (r"\b403\b", "forbidden", "Access denied"),
    (r"\b404\b", "not found", "Resource not found"),
    (r"\b405\b", "method not allowed", "HTTP method not supported"),
    (r"\b409\b", "conflict", "Resource conflict"),
    (r"\b422\b", "unprocessable", "Validation error"),
    # Schema/validation errors
    (r"schema", "schema error", "Invalid data schema"),
    (r"validation", "validation error", "Input validation failed"),
    (r"invalid.*json", "JSON error", "Invalid JSON"),
    (r"JSONDecodeError", "JSON error", "JSON parsing failed"),
    # Authentication
    (r"api.?key", "API key error", "Invalid or missing API key"),
    (r"token.*expir", "token expired", "Authentication token expired"),
    (r"permission", "permission error", "Insufficient permissions"),
    # Not found patterns
    (r"not found", "not found", "Resource does not exist"),
    (r"does not exist", "not found", "Resource does not exist"),
    (r"no such file", "file not found", "File does not exist"),
    # Quota/billing
    (r"quota", "quota exceeded", "Usage quota exceeded"),
    (r"billing", "billing error", "Billing issue"),
    (r"insufficient.*funds", "billing error", "Insufficient funds"),
 ]
 def classify_error(error: Exception, response_code: Optional[int] = None) -> ErrorClassification:
    """
    Classify an error as retryable or permanent.
    Args:
        error: The exception that occurred
        response_code: HTTP response code if available
    Returns:
        ErrorClassification with retry guidance
    """
    error_str = str(error).lower()
    error_type = type(error).__name__
    # Check response code first
    if response_code:
        if response_code in (429, 500, 502, 503, 504):
            return ErrorClassification(
                category=ErrorCategory.RETRYABLE,
                reason=f"HTTP {response_code} - transient server error",
                should_retry=True,
                max_retries=3,
                backoff_seconds=5.0 if response_code == 429 else 2.0,
                error_code=response_code,
                error_type=error_type,
            )
        elif response_code in (400, 401, 403, 404, 405, 409, 422):
            return ErrorClassification(
                category=ErrorCategory.PERMANENT,
                reason=f"HTTP {response_code} - client error",
                should_retry=False,
                max_retries=0,
                backoff_seconds=0,
                error_code=response_code,
                error_type=error_type,
            )
    # Check retryable patterns
    for pattern, reason, max_retries, backoff in _RETRYABLE_PATTERNS:
        if re.search(pattern, error_str, re.IGNORECASE):
            return ErrorClassification(
                category=ErrorCategory.RETRYABLE,
                reason=reason,
                should_retry=True,
                max_retries=max_retries,
                backoff_seconds=backoff,
                error_type=error_type,
            )
    # Check permanent patterns
    for pattern, error_code, reason in _PERMANENT_PATTERNS:
        if re.search(pattern, error_str, re.IGNORECASE):
            return ErrorClassification(
                category=ErrorCategory.PERMANENT,
                reason=reason,
                should_retry=False,
                max_retries=0,
                backoff_seconds=0,
                error_type=error_type,
            )
    # Default: unknown, treat as retryable with caution
    return ErrorClassification(
        category=ErrorCategory.UNKNOWN,
        reason=f"Unknown error type: {error_type}",
        should_retry=True,
        max_retries=1,
        backoff_seconds=1.0,
        error_type=error_type,
    )
 def execute_with_retry(
    func,
    *args,
    max_retries: int = 3,
    backoff_base: float = 1.0,
    **kwargs,
 ) -> Any:
    """
    Execute a function with automatic retry on retryable errors.
    Args:
        func: Function to execute
        *args: Function arguments
        max_retries: Maximum retry attempts
        backoff_base: Base backoff time in seconds
        **kwargs: Function keyword arguments
    Returns:
        Function result
    Raises:
        Exception: If permanent error or max retries exceeded
    """
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            last_error = e
            # Classify the error
            classification = classify_error(e)
            logger.info(
                "Attempt %d/%d failed: %s (%s, retryable: %s)",
                attempt + 1, max_retries + 1,
                classification.reason,
                classification.category.value,
                classification.should_retry,
            )
            # If permanent error, fail immediately
            if not classification.should_retry:
                logger.error("Permanent error: %s", classification.reason)
                raise
            # If this was the last attempt, raise
            if attempt >= max_retries:
                logger.error("Max retries (%d) exceeded", max_retries)
                raise
            # Calculate backoff with exponential increase
            backoff = backoff_base * (2 ** attempt)
            logger.info("Retrying in %.1fs...", backoff)
            time.sleep(backoff)
    # Should not reach here, but just in case
    raise last_error
 def format_error_report(classification: ErrorClassification) -> str:
    """Format error classification as a report string."""
    icon = "🔄" if classification.should_retry else "❌"
    return f"{icon} {classification.category.value}: {classification.reason}"
--- a/tools/session_search_tool.py
+++ b/tools/session_search_tool.py
@@ -394,23 +394,6 @@ def session_search(
            if len(seen_sessions) >= limit:
                break
        # RIDER: Reader-guided reranking — sort sessions by LLM answerability
        # This bridges the R@5 vs E2E accuracy gap by prioritizing passages
        # the LLM can actually answer from, not just keyword matches.
        try:
            from agent.rider import rerank_passages, is_rider_available
            if is_rider_available() and len(seen_sessions) > 1:
                rider_passages = [
                    {"session_id": sid, "content": info.get("snippet", ""), "rank": i + 1}
                    for i, (sid, info) in enumerate(seen_sessions.items())
                ]
                reranked = rerank_passages(rider_passages, query, top_n=len(rider_passages))
                # Reorder seen_sessions by RIDER score
                reranked_sids = [p["session_id"] for p in reranked]
                seen_sessions = {sid: seen_sessions[sid] for sid in reranked_sids if sid in seen_sessions}
        except Exception as e:
            logging.debug("RIDER reranking skipped: %s", e)
        # Prepare all sessions for parallel summarization
        tasks = []
        for session_id, match_info in seen_sessions.items():
--- a/tools/skill_deps.py
+++ b/tools/skill_deps.py
@@ -0,0 +1,185 @@
 """
 Skill Dependency Resolver — Auto-install missing dependencies
 Checks skill frontmatter for `requires` field and ensures
 dependencies are installed before loading the skill.
 Issue: #754
 """
 import importlib
 import logging
 import subprocess
 import sys
 from typing import Any, Dict, List, Optional, Tuple
 logger = logging.getLogger(__name__)
 def parse_requires(frontmatter: Dict[str, Any]) -> List[str]:
    """
    Parse the `requires` field from skill frontmatter.
    Supports:
    - requires: [package1, package2]
    - requires: package1
    - requires:
        - package1
        - package2
    """
    requires = frontmatter.get("requires", [])
    if isinstance(requires, str):
        return [requires]
    if isinstance(requires, list):
        return [str(r) for r in requires if r]
    return []
 def check_dependency(package: str) -> Tuple[bool, str]:
    """
    Check if a Python package is installed.
    Returns:
        Tuple of (is_installed, version_or_error)
    """
    # Handle pip package names (e.g., "matrix-nio[e2e]")
    import_name = package.split("[")[0].replace("-", "_")
    try:
        mod = importlib.import_module(import_name)
        version = getattr(mod, "__version__", "installed")
        return True, version
    except ImportError:
        return False, "not installed"
 def check_dependencies(requires: List[str]) -> Tuple[List[str], List[str]]:
    """
    Check which dependencies are missing.
    Returns:
        Tuple of (installed, missing)
    """
    installed = []
    missing = []
    for pkg in requires:
        is_installed, _ = check_dependency(pkg)
        if is_installed:
            installed.append(pkg)
        else:
            missing.append(pkg)
    return installed, missing
 def install_dependency(package: str, quiet: bool = False) -> Tuple[bool, str]:
    """
    Install a Python package via pip.
    Returns:
        Tuple of (success, output_or_error)
    """
    try:
        cmd = [sys.executable, "-m", "pip", "install", package]
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=120
        )
        if result.returncode == 0:
            if not quiet:
                logger.info("Installed %s", package)
            return True, result.stdout
        else:
            logger.error("Failed to install %s: %s", package, result.stderr)
            return False, result.stderr
    except subprocess.TimeoutExpired:
        return False, "Installation timed out"
    except Exception as e:
        return False, str(e)
 def resolve_dependencies(
    frontmatter: Dict[str, Any],
    auto_install: bool = False,
    quiet: bool = False
 ) -> Tuple[bool, List[str], List[str]]:
    """
    Resolve skill dependencies.
    Args:
        frontmatter: Skill frontmatter dict
        auto_install: If True, install missing deps without asking
        quiet: If True, suppress output
    Returns:
        Tuple of (all_satisfied, installed_now, still_missing)
    """
    requires = parse_requires(frontmatter)
    if not requires:
        return True, [], []
    installed, missing = check_dependencies(requires)
    if not missing:
        if not quiet:
            logger.debug("All dependencies satisfied: %s", installed)
        return True, [], []
    if not auto_install:
        if not quiet:
            logger.warning("Missing dependencies: %s", missing)
        return False, [], missing
    # Auto-install missing dependencies
    installed_now = []
    still_missing = []
    for pkg in missing:
        if not quiet:
            logger.info("Installing missing dependency: %s", pkg)
        success, output = install_dependency(pkg, quiet=quiet)
        if success:
            installed_now.append(pkg)
        else:
            still_missing.append(pkg)
            logger.error("Failed to install %s: %s", pkg, output[:200])
    all_satisfied = len(still_missing) == 0
    return all_satisfied, installed_now, still_missing
 def check_skill_dependencies(skill_dir) -> Dict[str, Any]:
    """
    Check dependencies for a skill directory.
    Returns:
        Dict with dependency status
    """
    from pathlib import Path
    skill_md = Path(skill_dir) / "SKILL.md"
    if not skill_md.exists():
        return {"requires": [], "installed": [], "missing": [], "satisfied": True}
    try:
        content = skill_md.read_text()
        from agent.skill_utils import parse_frontmatter
        frontmatter, _ = parse_frontmatter(content)
    except Exception:
        return {"requires": [], "installed": [], "missing": [], "satisfied": True}
    requires = parse_requires(frontmatter)
    installed, missing = check_dependencies(requires)
    return {
        "requires": requires,
        "installed": installed,
        "missing": missing,
        "satisfied": len(missing) == 0
    }
Author	SHA1	Message	Date
Alexander Whitestone	244dae106d	test: Add dependency resolver tests (#754 ) Some checks failed Docker Build and Publish / build-and-push (pull_request) Has been skipped Details Contributor Attribution Check / check-attribution (pull_request) Failing after 33s Details Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 27s Details Tests / e2e (pull_request) Successful in 3m19s Details Tests / test (pull_request) Failing after 59m22s Details	2026-04-15 03:46:10 +00:00
Alexander Whitestone	a606660c9d	feat: Add skill dependency resolver (#754 )	2026-04-15 03:45:53 +00:00