Compare commits

..

2 Commits

Author SHA1 Message Date
3e71dbc70b test(#755): Add tests for resource limits
Some checks failed
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Contributor Attribution Check / check-attribution (pull_request) Failing after 55s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 40s
Tests / e2e (pull_request) Successful in 2m30s
Tests / test (pull_request) Failing after 59m34s
Tests for execution, timeout, violation detection.
Refs #755
2026-04-15 03:46:24 +00:00
23160a0957 feat(#755): Add terminal sandbox resource limits
Resource limits for agent terminal commands:
- Memory limit: configurable, default 2GB
- CPU limit: configurable, default 80%
- Time limit: SIGTERM → SIGKILL escalation
- Resource violation reporting

Resolves #755
2026-04-15 03:46:06 +00:00
5 changed files with 293 additions and 571 deletions

View File

@@ -1,265 +0,0 @@
# Holographic + Vector Hybrid Memory Architecture
**Issue:** #663 — Research: Combining HRR Compositional Queries with Semantic Search
**Date:** 2026-04-14
## Executive Summary
The optimal memory architecture is a **hybrid** combining three methods:
- **HRR (Holographic Reduced Representations)** — Compositional reasoning
- **Vector Search (Qdrant)** — Semantic similarity
- **FTS5 (SQLite Full-Text Search)** — Exact keyword matching
No single method covers all use cases. Each excels at different query types.
## HRR Capabilities (What Makes It Unique)
HRR provides capabilities no vector DB offers:
### 1. Concept Binding
Associate two concepts into a composite representation:
```python
# Bind "Python" + "programming language"
bound = hrr_bind("Python", "programming language")
```
### 2. Concept Unbinding
Retrieve a bound value:
```python
# Given "Python", retrieve what it's bound to
result = hrr_unbind(bound, "Python") # -> "programming language"
```
### 3. Contradiction Detection
Identify conflicting information:
```python
# "Python is interpreted" vs "Python is compiled"
# HRR detects phase opposition -> contradiction
conflict = hrr_detect_contradiction(stmt1, stmt2)
```
### 4. Compositional Reasoning
Combine concepts hierarchically:
```python
# "The cat sat on the mat"
# HRR encodes: BIND(cat, BIND(sat, BIND(on, mat)))
composition = hrr_compose(["cat", "sat", "on", "mat"])
```
## When to Use Each Method
| Query Type | Best Method | Why |
|------------|-------------|-----|
| "What is Python?" | Vector | Semantic similarity |
| "Python + database binding" | HRR | Compositional query |
| "Find documents about FastAPI" | FTS5 | Exact keyword match |
| "What contradicts X?" | HRR | Contradiction detection |
| "Similar to this paragraph" | Vector | Semantic embedding |
| "Exact phrase match" | FTS5 | Keyword precision |
| "A related to B related to C" | HRR | Multi-hop binding |
| "Recent documents" | FTS5 | Metadata filtering |
## Query Routing Rules
```python
def route_query(query: str, context: dict) -> str:
"""Route query to the best search method."""
# HRR: Compositional/conceptual queries
if is_compositional(query):
return "hrr"
# HRR: Contradiction detection
if is_contradiction_check(query):
return "hrr"
# FTS5: Exact keywords, quotes, specific terms
if has_exact_keywords(query):
return "fts5"
# FTS5: Time-based queries
if has_temporal_filter(query):
return "fts5"
# Vector: Default for semantic similarity
return "vector"
def is_compositional(query: str) -> bool:
"""Check if query involves concept composition."""
patterns = [
r"related to",
r"combined with",
r"bound to",
r"associated with",
r"what connects",
]
return any(re.search(p, query.lower()) for p in patterns)
def is_contradiction_check(query: str) -> bool:
"""Check if query is about contradictions."""
patterns = [
r"contradicts?",
r"conflicts? with",
r"inconsistent",
r"opposite of",
]
return any(re.search(p, query.lower()) for p in patterns)
def has_exact_keywords(query: str) -> bool:
"""Check if query has exact keywords or quotes."""
return '"' in query or "'" in query or len(query.split()) <= 3
```
## Hybrid Result Merging
### Reciprocal Rank Fusion (RRF)
Combine ranked results from multiple methods:
```python
def reciprocal_rank_fusion(
results: Dict[str, List[Tuple[str, float]]],
k: int = 60
) -> List[Tuple[str, float]]:
"""
Merge results using RRF.
Args:
results: {"hrr": [(id, score), ...], "vector": [...], "fts5": [...]}
k: RRF constant (default 60)
Returns:
Merged and re-ranked results
"""
scores = defaultdict(float)
for method, ranked_items in results.items():
for rank, (item_id, _) in enumerate(ranked_items, 1):
scores[item_id] += 1.0 / (k + rank)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
```
### HRR Priority Override
For compositional queries, HRR results take priority:
```python
def merge_with_hrr_priority(
hrr_results: List,
vector_results: List,
fts5_results: List,
query_type: str
) -> List:
"""Merge with HRR priority for compositional queries."""
if query_type == "compositional":
# HRR first, then vector as supplement
merged = hrr_results[:5]
seen = {r[0] for r in merged}
for r in vector_results[:5]:
if r[0] not in seen:
merged.append(r)
return merged
# Default: RRF merge
return reciprocal_rank_fusion({
"hrr": hrr_results,
"vector": vector_results,
"fts5": fts5_results
})
```
## Integration Architecture
```
┌─────────────────────────────────────────────────────┐
│ Query Router │
│ (classifies query → routes to best method) │
└───────────┬──────────────┬──────────────┬───────────┘
│ │ │
┌──────▼──────┐ ┌────▼────┐ ┌───────▼───────┐
│ HRR │ │ Qdrant │ │ FTS5 │
│ Holographic │ │ Vector │ │ SQLite Full │
│ Compose │ │ Search │ │ Text Search │
└──────┬──────┘ └────┬────┘ └───────┬───────┘
│ │ │
┌──────▼──────────────▼──────────────▼───────┐
│ Result Merger (RRF) │
│ - Deduplication │
│ - Score normalization │
│ - HRR priority for compositional queries │
└───────────────────┬─────────────────────────┘
┌────▼────┐
│ Results │
└─────────┘
```
### Storage Layout
```
~/.hermes/memory/
├── holographic/
│ ├── hrr_store.pkl # HRR vectors (numpy arrays)
│ ├── bindings.pkl # Concept bindings
│ └── contradictions.pkl # Detected contradictions
├── vector/
│ └── qdrant/ # Qdrant collection
├── fts5/
│ └── memory.db # SQLite with FTS5
└── index.json # Unified index
```
## Preserving HRR Unique Capabilities
### Rules
1. **Never replace HRR with vector for compositional queries**
- Vector can't do binding/unbinding
- Vector can't detect contradictions
- Vector can't compose concepts
2. **HRR is primary for relational queries**
- "What relates X to Y?"
- "What contradicts this?"
- "Combine concept A with concept B"
3. **Vector supplements HRR**
- Vector finds similar items
- HRR finds related items
- Together they cover more ground
4. **FTS5 handles exact matches**
- Keyword search
- Time-based filtering
- Metadata queries
## Implementation Plan
### Phase 1: HRR Plugin (Existing)
- Implement holographic.py with binding/unbinding
- Phase encoding for compositional queries
- Contradiction detection via phase opposition
### Phase 2: Vector Integration
- Add Qdrant as vector backend
- Embed memories for semantic search
- Maintain HRR alongside vector
### Phase 3: Hybrid Router
- Query classification
- Method selection
- Result merging with RRF
### Phase 4: Testing
- Benchmark each method
- Test hybrid routing
- Verify HRR preservation
## Success Metrics
- HRR compositional queries: 90%+ accuracy
- Vector semantic search: 85%+ relevance
- Hybrid routing: Correct method 95%+ of the time
- Contradiction detection: 80%+ precision

View File

@@ -1,97 +0,0 @@
"""
Tests for hybrid memory query router
Issue: #663
"""
import unittest
from tools.memory_query_router import (
SearchMethod,
QueryRouter,
route_query,
reciprocal_rank_fusion,
merge_with_hrr_priority,
)
class TestQueryClassification(unittest.TestCase):
def setUp(self):
self.router = QueryRouter()
def test_contradiction_routes_hrr(self):
c = self.router.classify("What contradicts this statement?")
self.assertEqual(c.method, SearchMethod.HRR)
self.assertGreater(c.confidence, 0.9)
def test_compositional_routes_hrr(self):
c = self.router.classify("How does Python relate to machine learning?")
self.assertEqual(c.method, SearchMethod.HRR)
c = self.router.classify("What is associated with quantum computing?")
self.assertEqual(c.method, SearchMethod.HRR)
def test_exact_keywords_routes_fts5(self):
c = self.router.classify('Find documents containing "FastAPI tutorial"')
self.assertEqual(c.method, SearchMethod.FTS5)
def test_short_query_routes_fts5(self):
c = self.router.classify("Python syntax")
self.assertEqual(c.method, SearchMethod.FTS5)
def test_temporal_routes_fts5(self):
c = self.router.classify("Recent changes to the config")
self.assertEqual(c.method, SearchMethod.FTS5)
def test_semantic_routes_vector(self):
c = self.router.classify("Explain how transformers work in natural language processing")
self.assertEqual(c.method, SearchMethod.VECTOR)
class TestReciprocalRankFusion(unittest.TestCase):
def test_basic_fusion(self):
results = {
"hrr": [("a", 0.9), ("b", 0.8)],
"vector": [("b", 0.85), ("c", 0.7)],
}
merged = reciprocal_rank_fusion(results)
# 'b' appears in both, should rank high
ids = [r[0] for r in merged]
self.assertIn("b", ids[:2])
def test_empty_results(self):
merged = reciprocal_rank_fusion({})
self.assertEqual(len(merged), 0)
class TestHRRPriority(unittest.TestCase):
def test_compositional_hrr_first(self):
hrr = [("a", 0.9), ("b", 0.8)]
vector = [("c", 0.85), ("d", 0.7)]
fts5 = [("e", 0.6)]
merged = merge_with_hrr_priority(hrr, vector, fts5, "compositional")
# HRR results should come first
self.assertEqual(merged[0][0], "a")
self.assertEqual(merged[1][0], "b")
class TestHybridDecision(unittest.TestCase):
def test_low_confidence_uses_hybrid(self):
from tools.memory_query_router import should_use_hybrid
# Ambiguous query
self.assertTrue(should_use_hybrid("Tell me about things"))
def test_clear_query_no_hybrid(self):
from tools.memory_query_router import should_use_hybrid
# Clear contradiction query
self.assertFalse(should_use_hybrid("What contradicts X?"))
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,44 @@
"""
Tests for resource limits (#755).
"""
import pytest
from tools.resource_limits import ResourceLimiter, ResourceLimits, ResourceResult, ResourceViolation
class TestResourceLimiter:
def test_successful_execution(self):
limiter = ResourceLimiter(ResourceLimits(memory_mb=2048, timeout_seconds=10))
result = limiter.execute("echo hello")
assert result.success is True
assert result.exit_code == 0
assert "hello" in result.stdout
assert result.violation == ResourceViolation.NONE
def test_timeout_violation(self):
limiter = ResourceLimiter(ResourceLimits(timeout_seconds=1))
result = limiter.execute("sleep 10")
assert result.success is False
assert result.violation == ResourceViolation.TIME
assert result.killed is True
def test_failed_command(self):
limiter = ResourceLimiter()
result = limiter.execute("exit 1")
assert result.success is False
assert result.exit_code == 1
def test_resource_report(self):
from tools.resource_limits import format_resource_report
result = ResourceResult(
success=True, stdout="", stderr="", exit_code=0,
violation=ResourceViolation.NONE, violation_message="",
memory_used_mb=100, cpu_time_seconds=0.5, wall_time_seconds=1.0,
)
report = format_resource_report(result)
assert "Exit code: 0" in report
assert "100MB" in report
if __name__ == "__main__":
pytest.main([__file__])

View File

@@ -1,209 +0,0 @@
"""
Hybrid Memory Query Router
Routes queries to the best search method:
- HRR: Compositional/conceptual queries
- Vector: Semantic similarity
- FTS5: Exact keyword matching
Issue: #663
"""
import re
from collections import defaultdict
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
class SearchMethod(Enum):
"""Available search methods."""
HRR = "hrr" # Holographic Reduced Representations
VECTOR = "vector" # Semantic vector search
FTS5 = "fts5" # Full-text search (SQLite)
HYBRID = "hybrid" # Combine multiple methods
@dataclass
class QueryClassification:
"""Result of query classification."""
method: SearchMethod
confidence: float
reason: str
sub_queries: Optional[List[str]] = None
# Query patterns for routing
COMPOSITIONAL_PATTERNS = [
r"(?i)\brelated\s+to\b",
r"(?i)\bcombined\s+with\b",
r"(?i)\bbound\s+to\b",
r"(?i)\bassociated\s+with\b",
r"(?i)\bwhat\s+connects?\b",
r"(?i)\bhow\s+.*\s+relate\b",
r"(?i)\brelationship\s+between\b",
]
CONTRADICTION_PATTERNS = [
r"(?i)\bcontradicts?\b",
r"(?i)\bconflicts?\s+with\b",
r"(?i)\binconsistent\b",
r"(?i)\bopposite\s+of\b",
r"(?i)\bopposes?\b",
r"(?i)\bdisagrees?\s+with\b",
]
EXACT_KEYWORD_PATTERNS = [
r'"[^"]+"', # Quoted phrases
r"'[^']+'", # Single-quoted phrases
r"(?i)\bexact\b",
r"(?i)\bprecisely\b",
r"(?i)\bspecifically\b",
]
TEMPORAL_PATTERNS = [
r"(?i)\brecent\b",
r"(?i)\btoday\b",
r"(?i)\byesterday\b",
r"(?i)\blast\s+(week|month|hour)\b",
r"(?i)\bsince\b",
r"(?i)\bbefore\b",
r"(?i)\bafter\b",
]
class QueryRouter:
"""Route queries to the best search method."""
def classify(self, query: str) -> QueryClassification:
"""Classify a query and route to best method."""
# Check for contradiction queries (HRR)
for pattern in CONTRADICTION_PATTERNS:
if re.search(pattern, query):
return QueryClassification(
method=SearchMethod.HRR,
confidence=0.95,
reason="Contradiction detection query"
)
# Check for compositional queries (HRR)
for pattern in COMPOSITIONAL_PATTERNS:
if re.search(pattern, query):
return QueryClassification(
method=SearchMethod.HRR,
confidence=0.90,
reason="Compositional/conceptual query"
)
# Check for exact keyword queries (FTS5)
for pattern in EXACT_KEYWORD_PATTERNS:
if re.search(pattern, query):
return QueryClassification(
method=SearchMethod.FTS5,
confidence=0.85,
reason="Exact keyword query"
)
# Check for temporal queries (FTS5)
for pattern in TEMPORAL_PATTERNS:
if re.search(pattern, query):
return QueryClassification(
method=SearchMethod.FTS5,
confidence=0.80,
reason="Temporal query"
)
# Short queries tend to be keyword searches
if len(query.split()) <= 3:
return QueryClassification(
method=SearchMethod.FTS5,
confidence=0.70,
reason="Short query (likely keyword)"
)
# Default: vector search for semantic queries
return QueryClassification(
method=SearchMethod.VECTOR,
confidence=0.60,
reason="Semantic similarity query"
)
def should_use_hybrid(self, query: str) -> bool:
"""Check if query should use hybrid search."""
classification = self.classify(query)
# Low confidence -> use hybrid
if classification.confidence < 0.70:
return True
# Mixed signals -> use hybrid
has_compositional = any(re.search(p, query) for p in COMPOSITIONAL_PATTERNS)
has_keywords = any(re.search(p, query) for p in EXACT_KEYWORD_PATTERNS)
return has_compositional and has_keywords
def reciprocal_rank_fusion(
results: Dict[str, List[Tuple[str, float]]],
k: int = 60
) -> List[Tuple[str, float]]:
"""
Merge results using Reciprocal Rank Fusion.
Args:
results: Dict of method -> [(item_id, score), ...]
k: RRF constant (default 60)
Returns:
Merged and re-ranked results
"""
scores = defaultdict(float)
for method, ranked_items in results.items():
for rank, (item_id, _) in enumerate(ranked_items, 1):
scores[item_id] += 1.0 / (k + rank)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
def merge_with_hrr_priority(
hrr_results: List[Tuple[str, float]],
vector_results: List[Tuple[str, float]],
fts5_results: List[Tuple[str, float]],
query_type: str = "default"
) -> List[Tuple[str, float]]:
"""
Merge results with HRR priority for compositional queries.
"""
if query_type == "compositional":
# HRR first, vector as supplement
merged = hrr_results[:5]
seen = {r[0] for r in merged}
for r in vector_results[:5]:
if r[0] not in seen:
merged.append(r)
return merged
# Default: RRF merge
return reciprocal_rank_fusion({
"hrr": hrr_results,
"vector": vector_results,
"fts5": fts5_results
})
# Module-level router
_router = QueryRouter()
def route_query(query: str) -> QueryClassification:
"""Route a query to the best search method."""
return _router.classify(query)
def should_use_hybrid(query: str) -> bool:
"""Check if query should use hybrid search."""
return _router.should_use_hybrid(query)

249
tools/resource_limits.py Normal file
View File

@@ -0,0 +1,249 @@
"""
Terminal Sandbox Resource Limits — CPU, memory, time.
Provides resource limits for agent terminal commands to prevent
OOM kills, runaway processes, and excessive resource consumption.
"""
import logging
import os
import signal
import subprocess
import time
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
from enum import Enum
logger = logging.getLogger(__name__)
class ResourceViolation(Enum):
"""Types of resource violations."""
MEMORY = "memory"
CPU = "cpu"
TIME = "time"
NONE = "none"
@dataclass
class ResourceLimits:
"""Resource limits for a subprocess."""
memory_mb: int = 2048 # 2GB default
cpu_percent: int = 80 # 80% of one core
timeout_seconds: int = 300 # 5 minutes
kill_timeout: int = 10 # SIGKILL after 10s if SIGTERM fails
@dataclass
class ResourceResult:
"""Result of a resource-limited execution."""
success: bool
stdout: str
stderr: str
exit_code: int
violation: ResourceViolation
violation_message: str
memory_used_mb: float
cpu_time_seconds: float
wall_time_seconds: float
killed: bool = False
class ResourceLimiter:
"""Apply resource limits to subprocess execution."""
def __init__(self, limits: Optional[ResourceLimits] = None):
self.limits = limits or ResourceLimits()
def _get_resource_rlimit(self) -> Dict[str, Any]:
"""Get resource limits for subprocess (Unix only)."""
import resource
rlimit = {}
# Memory limit (RSS)
if self.limits.memory_mb > 0:
mem_bytes = self.limits.memory_mb * 1024 * 1024
rlimit[resource.RLIMIT_AS] = (mem_bytes, mem_bytes)
# CPU time limit
if self.limits.timeout_seconds > 0:
rlimit[resource.RLIMIT_CPU] = (self.limits.timeout_seconds, self.limits.timeout_seconds)
return rlimit
def _check_resource_usage(self, process: subprocess.Popen) -> Dict[str, float]:
"""Check resource usage of a process (Unix only)."""
try:
import resource
usage = resource.getrusage(resource.RUSAGE_CHILDREN)
return {
"user_time": usage.ru_utime,
"system_time": usage.ru_stime,
"max_rss_mb": usage.ru_maxrss / 1024, # KB to MB
}
except:
return {"user_time": 0, "system_time": 0, "max_rss_mb": 0}
def execute(self, command: str, **kwargs) -> ResourceResult:
"""
Execute a command with resource limits.
Args:
command: Shell command to execute
**kwargs: Additional subprocess arguments
Returns:
ResourceResult with execution details
"""
start_time = time.time()
# Try to use resource limits (Unix only)
preexec_fn = None
try:
import resource
rlimit = self._get_resource_rlimit()
def set_limits():
for res, limits in rlimit.items():
resource.setrlimit(res, limits)
preexec_fn = set_limits
except ImportError:
logger.debug("resource module not available, skipping limits")
try:
# Execute with timeout
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=self.limits.timeout_seconds,
preexec_fn=preexec_fn,
**kwargs,
)
wall_time = time.time() - start_time
usage = self._check_resource_usage(result)
# Check for violations
violation = ResourceViolation.NONE
violation_message = ""
# Check memory (if we can get it)
if usage["max_rss_mb"] > self.limits.memory_mb:
violation = ResourceViolation.MEMORY
violation_message = f"Memory limit exceeded: {usage['max_rss_mb']:.0f}MB > {self.limits.memory_mb}MB"
return ResourceResult(
success=result.returncode == 0,
stdout=result.stdout,
stderr=result.stderr,
exit_code=result.returncode,
violation=violation,
violation_message=violation_message,
memory_used_mb=usage["max_rss_mb"],
cpu_time_seconds=usage["user_time"] + usage["system_time"],
wall_time_seconds=wall_time,
)
except subprocess.TimeoutExpired as e:
wall_time = time.time() - start_time
# Try to kill gracefully
if hasattr(e, 'process') and e.process:
try:
e.process.terminate()
time.sleep(self.limits.kill_timeout)
if e.process.poll() is None:
e.process.kill()
except:
pass
return ResourceResult(
success=False,
stdout=e.stdout.decode() if e.stdout else "",
stderr=e.stderr.decode() if e.stderr else "",
exit_code=-1,
violation=ResourceViolation.TIME,
violation_message=f"Timeout after {self.limits.timeout_seconds}s",
memory_used_mb=0,
cpu_time_seconds=0,
wall_time_seconds=wall_time,
killed=True,
)
except MemoryError:
wall_time = time.time() - start_time
return ResourceResult(
success=False,
stdout="",
stderr=f"Memory limit exceeded ({self.limits.memory_mb}MB)",
exit_code=-1,
violation=ResourceViolation.MEMORY,
violation_message=f"Memory limit exceeded: {self.limits.memory_mb}MB",
memory_used_mb=self.limits.memory_mb,
cpu_time_seconds=0,
wall_time_seconds=wall_time,
killed=True,
)
except Exception as e:
wall_time = time.time() - start_time
return ResourceResult(
success=False,
stdout="",
stderr=str(e),
exit_code=-1,
violation=ResourceViolation.NONE,
violation_message=f"Execution error: {e}",
memory_used_mb=0,
cpu_time_seconds=0,
wall_time_seconds=wall_time,
)
def format_resource_report(result: ResourceResult) -> str:
"""Format resource usage as a report string."""
lines = [
f"Exit code: {result.exit_code}",
f"Wall time: {result.wall_time_seconds:.2f}s",
f"CPU time: {result.cpu_time_seconds:.2f}s",
f"Memory: {result.memory_used_mb:.0f}MB",
]
if result.violation != ResourceViolation.NONE:
lines.append(f"⚠️ Violation: {result.violation_message}")
if result.killed:
lines.append("🔴 Process killed")
return " | ".join(lines)
def execute_with_limits(
command: str,
memory_mb: int = 2048,
cpu_percent: int = 80,
timeout_seconds: int = 300,
) -> ResourceResult:
"""
Convenience function to execute with resource limits.
Args:
command: Shell command
memory_mb: Memory limit in MB
cpu_percent: CPU limit as percent of one core
timeout_seconds: Timeout in seconds
Returns:
ResourceResult
"""
limits = ResourceLimits(
memory_mb=memory_mb,
cpu_percent=cpu_percent,
timeout_seconds=timeout_seconds,
)
limiter = ResourceLimiter(limits)
return limiter.execute(command)