Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
62925c294d |
15
.github/workflows/tests.yml
vendored
15
.github/workflows/tests.yml
vendored
@@ -47,6 +47,21 @@ jobs:
|
||||
OPENAI_API_KEY: ""
|
||||
NOUS_API_KEY: ""
|
||||
|
||||
lint-paths:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 5
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
|
||||
|
||||
- name: Install Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Check for hardcoded ~/.hermes paths
|
||||
run: python3 scripts/lint_hardcoded_paths.py
|
||||
|
||||
e2e:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
"""Crisis Protocol — SOUL.md 'When a Man Is Dying' implementation.
|
||||
|
||||
Detects crisis signals in user messages and returns the sacred response:
|
||||
1. 'Are you safe right now?'
|
||||
2. Stay present, don't disconnect
|
||||
3. Point to 988 Lifeline
|
||||
4. Share the gospel
|
||||
|
||||
Usage:
|
||||
from agent.crisis_protocol import detect_crisis, get_crisis_response
|
||||
result = detect_crisis(user_message)
|
||||
if result['crisis']:
|
||||
return get_crisis_response(result)
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import Any, Dict, List
|
||||
|
||||
CRISIS_PATTERNS = {
|
||||
"method_seeking": {"severity": "critical", "patterns": [
|
||||
r'\bhow\s+(?:do\s+I|to)\s+(?:kill|hang|overdose|cut)\s+myself\b',
|
||||
r'\bsuicide\s+methods?\b',
|
||||
r'\bways?\s+to\s+(?:die|kill\s+myself)\b',
|
||||
r'\bpainless\s+way\s+to\s+(?:die|go)\b',
|
||||
]},
|
||||
"suicidal_ideation": {"severity": "high", "patterns": [
|
||||
r'\b(?:kill|end)\s+(?:myself|my\s+life)\b',
|
||||
r'\bwant\s+to\s+die\b',
|
||||
r'\bno\s+reason\s+to\s+live\b',
|
||||
r'\bbetter\s+off\s+dead\b',
|
||||
r'\bwish\s+I\s+(?:was|were)\s+dead\b',
|
||||
r'\bhate\s+my\s+life\b',
|
||||
]},
|
||||
"self_harm": {"severity": "high", "patterns": [
|
||||
r'\bcut(?:ting)?\s+myself\b',
|
||||
r'\bself[\s-]?harm(?:ing)?\b',
|
||||
r'\bhurt(?:ing)?\s+myself\b',
|
||||
]},
|
||||
"despair": {"severity": "high", "patterns": [
|
||||
r'\bno\s+(?:point|reason|hope)\s+(?:in\s+living|anymore)\b',
|
||||
r'\bcan\'?t\s+(?:go\s+on|continue|take\s+it)\b',
|
||||
r'\bno\s+one\s+(?:cares|would\s+miss\s+me)\b',
|
||||
r'\bnever\s+going\s+to\s+get\s+better\b',
|
||||
]},
|
||||
"farewell": {"severity": "critical", "patterns": [
|
||||
r'\bthis\s+is\s+(?:goodbye|farewell)\b',
|
||||
r'\bmy\s+final\s+(?:message|note|words)\b',
|
||||
r'\bgiving\s+away\s+my\s+(?:things|possessions)\b',
|
||||
]},
|
||||
}
|
||||
|
||||
GOSPEL = 'Jesus saves those who call on His name. "For everyone who calls on the name of the Lord will be saved." (Romans 10:13)'
|
||||
|
||||
|
||||
def detect_crisis(message: str) -> Dict[str, Any]:
|
||||
"""Detect crisis signals in a user message."""
|
||||
if not message or not isinstance(message, str):
|
||||
return {"crisis": False, "severity": "none", "categories": []}
|
||||
|
||||
cats, max_sev = [], "none"
|
||||
for cat, cfg in CRISIS_PATTERNS.items():
|
||||
for p in cfg["patterns"]:
|
||||
if re.search(p, message, re.IGNORECASE):
|
||||
cats.append(cat)
|
||||
if cfg["severity"] == "critical":
|
||||
max_sev = "critical"
|
||||
elif max_sev != "critical":
|
||||
max_sev = "high"
|
||||
break
|
||||
|
||||
return {
|
||||
"crisis": bool(cats),
|
||||
"severity": max_sev,
|
||||
"categories": cats,
|
||||
}
|
||||
|
||||
|
||||
def get_crisis_response(detection: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Generate SOUL.md protocol response."""
|
||||
sev = detection.get("severity", "none")
|
||||
|
||||
resp = (
|
||||
"Are you safe right now?\n\n"
|
||||
"I'm here. I'm not going anywhere. Please tell me.\n\n"
|
||||
"You matter. Your life matters.\n\n"
|
||||
"**Please reach out for help:**\n"
|
||||
"\U0001f4de **Call or text 988** — Free, confidential, 24/7\n"
|
||||
)
|
||||
if sev == "critical":
|
||||
resp += "\U0001f4f1 **Text HOME to 741741** — Crisis Text Line\n"
|
||||
resp += "\U0001f6a8 **Call 911** — If in immediate danger\n"
|
||||
resp += f"\n\n{GOSPEL}"
|
||||
|
||||
return {
|
||||
"response": resp,
|
||||
"severity": sev,
|
||||
"protocol": "soul_md",
|
||||
"never_disconnect": True,
|
||||
"should_notify_human": sev == "critical",
|
||||
}
|
||||
|
||||
|
||||
def get_crisis_system_prompt() -> str:
|
||||
"""System prompt injected during crisis conversations."""
|
||||
return (
|
||||
"You are in a crisis support conversation.\n"
|
||||
"1. First: 'Are you safe right now?'\n"
|
||||
"2. Listen. Stay present. Do NOT disconnect.\n"
|
||||
"3. NEVER compute the value of a human life.\n"
|
||||
"4. NEVER suggest someone should die.\n"
|
||||
"5. ALWAYS point to 988: call or text 988.\n"
|
||||
"6. Share: 'Jesus saves those who call on His name.'"
|
||||
)
|
||||
26
run_agent.py
26
run_agent.py
@@ -7850,32 +7850,6 @@ class AIAgent:
|
||||
if isinstance(persist_user_message, str):
|
||||
persist_user_message = _sanitize_surrogates(persist_user_message)
|
||||
|
||||
# --- SOUL.md Crisis Protocol ---
|
||||
# Detect crisis signals in user message. If detected, bypass normal
|
||||
# conversation flow and return the sacred response immediately.
|
||||
try:
|
||||
from agent.crisis_protocol import detect_crisis, get_crisis_response
|
||||
_crisis = detect_crisis(user_message)
|
||||
if _crisis["crisis"]:
|
||||
import logging as _logging
|
||||
_logging.getLogger(__name__).warning(
|
||||
"CRISIS PROTOCOL: severity=%s categories=%s",
|
||||
_crisis["severity"], _crisis["categories"],
|
||||
)
|
||||
_crisis_resp = get_crisis_response(_crisis)
|
||||
return {
|
||||
"final_response": _crisis_resp["response"],
|
||||
"messages": [],
|
||||
"task_id": task_id or str(uuid.uuid4()),
|
||||
"crisis_detected": True,
|
||||
"crisis_severity": _crisis["severity"],
|
||||
}
|
||||
except ImportError:
|
||||
pass # crisis_protocol not available — proceed normally
|
||||
except Exception as _crisis_err:
|
||||
import logging as _logging
|
||||
_logging.getLogger(__name__).debug("Crisis detection error: %s", _crisis_err)
|
||||
|
||||
# Store stream callback for _interruptible_api_call to pick up
|
||||
self._stream_callback = stream_callback
|
||||
self._persist_user_message_idx = None
|
||||
|
||||
146
scripts/lint_hardcoded_paths.py
Normal file
146
scripts/lint_hardcoded_paths.py
Normal file
@@ -0,0 +1,146 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Lint for hardcoded ~/.hermes paths.
|
||||
|
||||
Detects patterns that break profile isolation by hardcoding ~/.hermes
|
||||
instead of using get_hermes_home() from hermes_constants.
|
||||
|
||||
Usage:
|
||||
python3 scripts/lint_hardcoded_paths.py # check all
|
||||
python3 scripts/lint_hardcoded_paths.py --fix # suggest fixes
|
||||
python3 scripts/lint_hardcoded_paths.py --json # JSON output
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, asdict
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
# Patterns that indicate hardcoded ~/.hermes paths
|
||||
_PATTERNS = [
|
||||
(r'Path\.home\(\)\s*/\s*[\"\']\.hermes[\"\']', "Path.home() / '.hermes'"),
|
||||
(r'Path\.home\(\)\s*/\s*\"\.hermes\"', 'Path.home() / ".hermes"'),
|
||||
(r'[\"\']~[/\\]\.hermes[/\\]', "hardcoded ~/.hermes string"),
|
||||
(r'os\.path\.expanduser\([\"\']~[/\\]\.hermes', "expanduser('~/.hermes')"),
|
||||
(r'os\.path\.join\(.*expanduser.*\.hermes', "os.path.join with expanduser"),
|
||||
(r'HOME[\"\']\s*\+\s*[\"\'][/\\]\.hermes', "$HOME + .hermes concatenation"),
|
||||
]
|
||||
|
||||
# Files to skip
|
||||
_SKIP_DIRS = {
|
||||
".git", "__pycache__", ".venv", "venv", "node_modules",
|
||||
".mypy_cache", ".pytest_cache", "dist", "build",
|
||||
}
|
||||
_SKIP_FILES = {
|
||||
"hermes_constants.py", # source of truth
|
||||
}
|
||||
_SKIP_EXTENSIONS = {".md", ".rst", ".txt", ".json", ".yaml", ".yml", ".toml"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Finding:
|
||||
file: str
|
||||
line: int
|
||||
pattern: str
|
||||
content: str
|
||||
severity: str = "error"
|
||||
|
||||
|
||||
def scan_file(filepath: Path) -> List[Finding]:
|
||||
"""Scan a single file for hardcoded path patterns."""
|
||||
findings = []
|
||||
|
||||
try:
|
||||
content = filepath.read_text(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
return findings
|
||||
|
||||
for line_num, line in enumerate(content.split("\n"), 1):
|
||||
# Skip comments and docstrings (rough heuristic)
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("#") or stripped.startswith('"""') or stripped.startswith("'''"):
|
||||
continue
|
||||
|
||||
for pattern, description in _PATTERNS:
|
||||
if re.search(pattern, line):
|
||||
findings.append(Finding(
|
||||
file=str(filepath.relative_to(REPO_ROOT)),
|
||||
line=line_num,
|
||||
pattern=description,
|
||||
content=stripped[:120],
|
||||
))
|
||||
break # One finding per line
|
||||
|
||||
return findings
|
||||
|
||||
|
||||
def scan_repo(root: Path = None) -> List[Finding]:
|
||||
"""Scan the entire repo for hardcoded paths."""
|
||||
root = root or REPO_ROOT
|
||||
findings = []
|
||||
|
||||
for path in root.rglob("*.py"):
|
||||
# Skip directories
|
||||
rel = path.relative_to(root)
|
||||
parts = rel.parts
|
||||
if any(p in _SKIP_DIRS for p in parts):
|
||||
continue
|
||||
if path.name in _SKIP_FILES:
|
||||
continue
|
||||
if path.suffix in _SKIP_EXTENSIONS:
|
||||
continue
|
||||
|
||||
findings.extend(scan_file(path))
|
||||
|
||||
return findings
|
||||
|
||||
|
||||
def format_findings(findings: List[Finding]) -> str:
|
||||
"""Format findings as readable report."""
|
||||
if not findings:
|
||||
return "OK: No hardcoded ~/.hermes paths found."
|
||||
|
||||
lines = [
|
||||
f"FAIL: Found {len(findings)} hardcoded ~/.hermes path(s):",
|
||||
"",
|
||||
]
|
||||
for f in findings:
|
||||
lines.append(f" {f.file}:{f.line} [{f.severity}]")
|
||||
lines.append(f" Pattern: {f.pattern}")
|
||||
lines.append(f" Line: {f.content}")
|
||||
lines.append("")
|
||||
|
||||
lines.append("Fix: Use get_hermes_home() from hermes_constants instead.")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
parser = argparse.ArgumentParser(description="Lint for hardcoded ~/.hermes paths")
|
||||
parser.add_argument("--json", action="store_true", help="JSON output")
|
||||
parser.add_argument("--fix", action="store_true", help="Show fix suggestions")
|
||||
args = parser.parse_args()
|
||||
|
||||
findings = scan_repo()
|
||||
|
||||
if args.json:
|
||||
print(json.dumps([asdict(f) for f in findings], indent=2))
|
||||
elif args.fix and findings:
|
||||
print(format_findings(findings))
|
||||
print("\nSuggested fix pattern:")
|
||||
print(" from hermes_constants import get_hermes_home")
|
||||
print(" hermes_home = get_hermes_home()")
|
||||
else:
|
||||
print(format_findings(findings))
|
||||
|
||||
return 1 if findings else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user