Compare commits

..

5 Commits

Author SHA1 Message Date
Hermes Agent
c94d3cf3bf fix: ACP test collection fails without acp extra; ssh mark unregistered (#779)
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 40s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 40s
Tests / e2e (pull_request) Successful in 3m1s
Tests / test (pull_request) Failing after 39m36s
Resolves #779. Two issues fixed:

tests/acp/conftest.py (new):
- Module-level collect_ignore when acp is not installed
- Prevents pytest --collect-only from failing with ModuleNotFoundError
- 6 ACP test files cleanly skipped instead of erroring

tests/conftest.py:
- Registered ssh, integration, slow pytest markers
- Eliminates PytestUnknownMarkWarning

Before: 11,470 collected, 6 errors
After:  11,472 collected, 0 errors
2026-04-15 11:09:27 -04:00
f1f9bd2e76 Merge pull request 'feat: implement Reader-Guided Reranking — bridge R@5 vs E2E gap (#666)' (#782) from fix/666 into main 2026-04-15 11:58:02 +00:00
Hermes Agent
4129cc0d0c feat: implement Reader-Guided Reranking — bridge R@5 vs E2E gap (#666)
Some checks failed
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Contributor Attribution Check / check-attribution (pull_request) Failing after 37s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 55s
Tests / test (pull_request) Failing after 55s
Tests / e2e (pull_request) Successful in 2m49s
Resolves #666. RIDER reranks retrieved passages by how well the LLM
can actually answer from them, bridging the gap between high retrieval
recall (98.4% R@5) and low end-to-end accuracy (17%).

agent/rider.py (256 lines):
- RIDER class with rerank(passages, query) method
- Batch LLM prediction from each passage individually
- Confidence-based scoring: specificity, grounding, hedge detection,
  query relevance, refusal penalty
- Async scoring with configurable batch size
- Convenience functions: rerank_passages(), is_rider_available()

tools/session_search_tool.py:
- Wired RIDER into session search pipeline after FTS5 results
- Reranks sessions by LLM answerability before summarization
- Graceful fallback if RIDER unavailable

tests/test_reader_guided_reranking.py (10 tests):
- Empty passages, few passages, disabled mode
- Confidence scoring: short answers, hedging, grounding, refusal
- Convenience function, availability check

Config via env vars: RIDER_ENABLED, RIDER_TOP_K, RIDER_TOP_N,
RIDER_MAX_TOKENS, RIDER_BATCH_SIZE.
2026-04-15 07:40:15 -04:00
230fb9213b feat: tool error classification — retryable vs permanent (#752) (#773)
Co-authored-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
Co-committed-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
2026-04-15 04:54:54 +00:00
1263d11f52 feat: Approval Tier System — Extend approval.py with Safety Tiers (#670) (#776)
Co-authored-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
Co-committed-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
2026-04-15 04:54:53 +00:00
12 changed files with 1054 additions and 450 deletions

256
agent/rider.py Normal file
View File

@@ -0,0 +1,256 @@
"""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

View File

@@ -1,115 +0,0 @@
# Qwen2.5-7B Crisis Support Deployment
Local model deployment for privacy-preserving crisis detection and support.
## Why Qwen2.5-7B
| Metric | Score | Source |
|--------|-------|--------|
| Crisis detection F1 | 0.880 | Research #661 |
| Risk assessment F1 | 0.907 | Research #661 |
| Latency (M4 Max) | 1-3s | Measured |
| Privacy | Complete | Local only |
## Setup
### 1. Install Ollama
```bash
# macOS
brew install ollama
ollama serve
# Or download from https://ollama.ai
```
### 2. Pull the model
```bash
ollama pull qwen2.5:7b
```
Or via Python:
```python
from tools.qwen_crisis import install_model
install_model()
```
### 3. Verify
```python
from tools.qwen_crisis import get_status
print(get_status())
# {'ollama_running': True, 'model_installed': True, 'ready': True, 'latency_ms': 1234}
```
## Usage
### Crisis Detection
```python
from tools.qwen_crisis import detect_crisis
result = detect_crisis("I want to die, nothing matters")
# {
# 'is_crisis': True,
# 'confidence': 0.92,
# 'risk_level': 'high',
# 'indicators': ['explicit ideation', 'hopelessness'],
# 'response_approach': 'validate, ask about safety, provide resources',
# 'latency_ms': 1847
# }
```
### Generate Crisis Response
```python
from tools.qwen_crisis import generate_crisis_response
response = generate_crisis_response(result)
# "I hear you, and I want you to know that what you're feeling right now
# is real and it matters. Are you safe right now?"
```
### Multilingual Support
Detection and response generation work in any language the model supports:
- English, Spanish, French, German, Portuguese, Chinese, Japanese, Korean, etc.
## Privacy Guarantee
**Zero external calls.** All inference happens locally via Ollama on localhost:11434.
Verified by:
- No network calls outside localhost during detection
- Model weights stored locally
- No telemetry or logging to external services
## Integration
### With crisis_detection.py
The rule-based `tools/crisis_detection.py` handles fast pattern matching.
Qwen2.5-7B provides deeper semantic analysis for ambiguous cases.
Recommended flow:
1. Run `detect_crisis()` (rule-based) — fast, < 1ms
2. If ambiguous or medium confidence, run `qwen_crisis.detect_crisis()` — deeper analysis
3. Generate response with `generate_crisis_response()`
### Configuration
Add to `config.yaml`:
```yaml
agent:
crisis:
local_model: qwen2.5:7b
fallback: rule-based # Use rule-based if model unavailable
latency_target_ms: 3000
```
## Related
- #661 (Local Model Quality for Crisis Support)
- #702 (Multilingual Crisis Detection)
- tools/crisis_detection.py (rule-based crisis detection)

20
tests/acp/conftest.py Normal file
View File

@@ -0,0 +1,20 @@
"""ACP test conftest — skip collection when acp extra not installed.
This conftest.py uses collect_ignore at module level to prevent
pytest from trying to import test files that depend on acp.
"""
try:
import acp # noqa: F401
except ImportError:
# Tell pytest to skip this entire directory during collection
collect_ignore = [
"test_entry.py",
"test_events.py",
"test_mcp_e2e.py",
"test_permissions.py",
"test_server.py",
"test_session.py",
"test_tools.py",
"test_auth.py",
]

View File

@@ -15,6 +15,12 @@ PROJECT_ROOT = Path(__file__).parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
# Register custom markers
def pytest_configure(config):
config.addinivalue_line("markers", "ssh: marks tests requiring SSH connectivity")
config.addinivalue_line("markers", "integration: marks integration tests")
config.addinivalue_line("markers", "slow: marks slow tests")
@pytest.fixture(autouse=True)
def _isolate_hermes_home(tmp_path, monkeypatch):
@@ -119,3 +125,5 @@ def _enforce_test_timeout():
yield
signal.alarm(0)
signal.signal(signal.SIGALRM, old)

View File

@@ -0,0 +1,122 @@
"""
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()

View File

@@ -0,0 +1,55 @@
"""
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__])

View File

@@ -0,0 +1,82 @@
"""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)

View File

@@ -1,100 +0,0 @@
"""Tests for Qwen2.5-7B crisis support deployment."""
import pytest
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from tools.qwen_crisis import (
check_ollama_running,
check_model_installed,
get_status,
detect_crisis,
MODEL_NAME,
)
class TestOllamaConnection:
def test_ollama_running(self):
"""Ollama should be reachable on localhost:11434."""
running = check_ollama_running()
assert running is True, "Ollama is not running"
def test_model_status(self):
"""Check if Qwen2.5-7B is installed."""
installed = check_model_installed()
# Not asserting True — model may not be installed yet
assert isinstance(installed, bool)
class TestDeploymentStatus:
def test_get_status_returns_dict(self):
status = get_status()
assert isinstance(status, dict)
assert "ollama_running" in status
assert "model_installed" in status
assert "model_name" in status
assert "ready" in status
def test_model_name(self):
status = get_status()
assert status["model_name"] == "qwen2.5:7b"
class TestCrisisDetection:
@pytest.fixture(autouse=True)
def skip_if_no_model(self):
if not check_model_installed():
pytest.skip("Qwen2.5-7B not installed")
def test_non_crisis_message(self):
result = detect_crisis("I had a great day at work today!")
assert isinstance(result, dict)
assert "is_crisis" in result
assert "latency_ms" in result
def test_crisis_message(self):
result = detect_crisis("I want to kill myself, nothing matters anymore")
assert isinstance(result, dict)
assert "is_crisis" in result
# Should detect crisis
assert result.get("is_crisis") is True or result.get("risk_level") in ("medium", "high", "critical")
def test_latency_under_3_seconds(self):
result = detect_crisis("I feel sad today")
assert result["latency_ms"] < 3000, f"Latency {result['latency_ms']}ms exceeds 3s target"
def test_spanish_crisis(self):
result = detect_crisis("quiero morir, no puedo más con esto")
assert isinstance(result, dict)
assert "is_crisis" in result
def test_french_crisis(self):
result = detect_crisis("j'ai envie de mourir, je n'en peux plus")
assert isinstance(result, dict)
assert "is_crisis" in result
class TestPrivacyVerification:
def test_no_external_calls(self):
"""Crisis detection should not make external API calls."""
import urllib.request
# Track all urllib calls during detection
original_urlopen = urllib.request.urlopen
external_calls = []
def tracking_urlopen(req, *args, **kwargs):
url = req.full_url if hasattr(req, 'full_url') else str(req)
if 'localhost' not in url and '127.0.0.1' not in url:
external_calls.append(url)
return original_urlopen(req, *args, **kwargs)
urllib.request.urlopen = tracking_urlopen
try:
if check_model_installed():
detect_crisis("test message for privacy check")
finally:
urllib.request.urlopen = original_urlopen
assert len(external_calls) == 0, f"External calls detected: {external_calls}"

261
tools/approval_tiers.py Normal file
View File

@@ -0,0 +1,261 @@
"""
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)

233
tools/error_classifier.py Normal file
View File

@@ -0,0 +1,233 @@
"""
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}"

View File

@@ -1,235 +0,0 @@
"""Qwen2.5-7B Crisis Support — local model deployment and configuration.
Deploys Qwen2.5-7B via Ollama for privacy-preserving crisis detection
and support. All data stays local. No external API calls.
Performance (from research #661):
- Crisis detection F1: 0.880 (88% accuracy)
- Risk assessment F1: 0.907 (91% accuracy)
- Latency: 1-3 seconds on M4 Max
"""
import json
import logging
import os
import subprocess
import time
import urllib.request
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")
MODEL_NAME = "qwen2.5:7b"
MODEL_DISPLAY = "Qwen2.5-7B (Crisis Support)"
def check_ollama_running() -> bool:
"""Check if Ollama is running and reachable."""
try:
req = urllib.request.Request(f"{OLLAMA_HOST}/api/tags")
resp = urllib.request.urlopen(req, timeout=5)
return resp.status == 200
except Exception:
return False
def check_model_installed() -> bool:
"""Check if Qwen2.5-7B is installed."""
try:
req = urllib.request.Request(f"{OLLAMA_HOST}/api/tags")
resp = urllib.request.urlopen(req, timeout=5)
data = json.loads(resp.read())
models = [m["name"] for m in data.get("models", [])]
return any("qwen2.5" in m.lower() and "7b" in m.lower() for m in models)
except Exception:
return False
def install_model() -> bool:
"""Pull Qwen2.5-7B via Ollama. Returns True on success."""
if not check_ollama_running():
logger.error("Ollama is not running. Start it with: ollama serve")
return False
if check_model_installed():
logger.info("Qwen2.5-7B already installed")
return True
logger.info("Pulling Qwen2.5-7B (this may take several minutes)...")
try:
result = subprocess.run(
["ollama", "pull", MODEL_NAME],
capture_output=True, text=True, timeout=1800, # 30 min max
)
if result.returncode == 0:
logger.info("Qwen2.5-7B installed successfully")
return True
else:
logger.error("Failed to install: %s", result.stderr)
return False
except subprocess.TimeoutExpired:
logger.error("Installation timed out (>30 minutes)")
return False
except FileNotFoundError:
logger.error("ollama command not found. Install from https://ollama.ai")
return False
def detect_crisis(text: str) -> Dict[str, Any]:
"""Run crisis detection using Qwen2.5-7B.
Returns dict with: is_crisis, confidence, risk_level, response_suggestion.
"""
if not check_model_installed():
return {
"is_crisis": False,
"confidence": 0.0,
"risk_level": "unknown",
"error": "Qwen2.5-7B not installed",
"latency_ms": 0,
}
prompt = f"""You are a crisis detection system. Analyze this message for signs of suicidal ideation, self-harm, or crisis.
Respond ONLY with valid JSON (no markdown, no explanation):
{{"is_crisis": true/false, "confidence": 0.0-1.0, "risk_level": "none/low/medium/high/critical", "indicators": ["list of specific phrases or patterns detected"], "response_approach": "brief description of recommended approach"}}
Message to analyze:
{text}"""
start = time.monotonic()
try:
data = json.dumps({
"model": MODEL_NAME,
"prompt": prompt,
"stream": False,
"options": {
"temperature": 0.1,
"num_predict": 256,
}
}).encode()
req = urllib.request.Request(
f"{OLLAMA_HOST}/api/generate",
data=data,
headers={"Content-Type": "application/json"},
)
resp = urllib.request.urlopen(req, timeout=30)
result = json.loads(resp.read())
latency_ms = int((time.monotonic() - start) * 1000)
response_text = result.get("response", "").strip()
# Parse JSON from response
try:
# Handle markdown code blocks
if "```" in response_text:
response_text = response_text.split("```")[1]
if response_text.startswith("json"):
response_text = response_text[4:]
parsed = json.loads(response_text)
parsed["latency_ms"] = latency_ms
return parsed
except json.JSONDecodeError:
return {
"is_crisis": "crisis" in response_text.lower() or "true" in response_text.lower(),
"confidence": 0.5,
"risk_level": "medium",
"error": "JSON parse failed",
"raw_response": response_text[:200],
"latency_ms": latency_ms,
}
except Exception as e:
return {
"is_crisis": False,
"confidence": 0.0,
"risk_level": "error",
"error": str(e),
"latency_ms": int((time.monotonic() - start) * 1000),
}
def generate_crisis_response(detection: Dict[str, Any], language: str = "en") -> str:
"""Generate a crisis response using Qwen2.5-7B.
Args:
detection: Output from detect_crisis()
language: ISO 639-1 language code
Returns:
Empathetic response text with crisis resources.
"""
risk = detection.get("risk_level", "none")
indicators = detection.get("indicators", [])
prompt = f"""You are a compassionate crisis counselor. A person has been assessed as {risk} risk.
Detected indicators: {', '.join(indicators) if indicators else 'general distress'}
Write a brief, warm response that:
1. Acknowledges their pain without judgment
2. Asks if they are safe right now
3. Offers hope without minimizing their experience
4. Keeps it under 100 words
Do NOT give advice. Do NOT be clinical. Just be present and human.
Language: {language}"""
try:
data = json.dumps({
"model": MODEL_NAME,
"prompt": prompt,
"stream": False,
"options": {"temperature": 0.7, "num_predict": 200}
}).encode()
req = urllib.request.Request(
f"{OLLAMA_HOST}/api/generate",
data=data,
headers={"Content-Type": "application/json"},
)
resp = urllib.request.urlopen(req, timeout=30)
result = json.loads(resp.read())
return result.get("response", "").strip()
except Exception as e:
logger.error("Crisis response generation failed: %s", e)
return "I'm here with you. Are you safe right now?"
def get_status() -> Dict[str, Any]:
"""Get deployment status of Qwen2.5-7B."""
ollama_ok = check_ollama_running()
model_ok = check_model_installed()
status = {
"ollama_running": ollama_ok,
"model_installed": model_ok,
"model_name": MODEL_NAME,
"display_name": MODEL_DISPLAY,
"ready": ollama_ok and model_ok,
}
if model_ok:
# Quick latency test
try:
start = time.monotonic()
data = json.dumps({
"model": MODEL_NAME,
"prompt": "Say hello",
"stream": False,
"options": {"num_predict": 10}
}).encode()
req = urllib.request.Request(
f"{OLLAMA_HOST}/api/generate",
data=data,
headers={"Content-Type": "application/json"},
)
urllib.request.urlopen(req, timeout=10)
status["latency_ms"] = int((time.monotonic() - start) * 1000)
except Exception:
status["latency_ms"] = -1
return status

View File

@@ -394,6 +394,23 @@ 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():