feat: add advisory action risk check skill

docs/action-risk-check.md

@@ -0,0 +1,56 @@
+# Action Risk Gate
+
+## Purpose
+
+`action-risk-check` deterministically evaluates the risk of an intended
+OmegaClaw action. It uses local rules only: no LLM, network request, command
+execution, or filesystem access occurs during classification. Results include
+a decision, risk level, and stable reason code, and never echo the supplied
+target.
+
+## Skill signature
+
+```metta
+(action-risk-check "skill-name" "target-or-argument")
+```
+
+The result follows the existing Python bridge convention and is returned as a
+plain string:
+
+```text
+decision=allow, risk=low, reason=read-only
+```
+
+## Examples
+
+```metta
+(action-risk-check "websearch" "public Hyperon documentation")
+; decision=allow, risk=low, reason=read-only
+
+(action-risk-check "write-file" "notes.txt")
+; decision=review, risk=medium, reason=state-changing
+
+(action-risk-check "read-file" "/project/.env")
+; decision=review, risk=high, reason=sensitive-target
+
+(action-risk-check "shell" "rm -rf /var/lib/app")
+; decision=block, risk=critical, reason=destructive-action
+```
+
+Classification precedence is fixed: destructive, sensitive, state-changing,
+read-only, then unknown. Unknown skills default to `review`/`medium`.
+
+## Advisory limitation
+
+This first version is advisory. It evaluates an action only when
+`action-risk-check` is explicitly called. It does not intercept every skill or
+prevent another skill from executing.
+
+## Possible Phase 2 enforcement
+
+A later pre-dispatch gate could classify each parsed skill expression in
+`src/loop.metta` before `eval`. It could permit `allow`, require explicit user
+confirmation for `review`, and refuse `block`, while keeping the classifier
+side-effect-free. Such enforcement should include a trusted confirmation
+state, protection against recursive gating, structured audit logging that
+redacts arguments, and regression tests for multi-command dispatch.

lib_omegaclaw.metta

@@ -21,6 +21,7 @@
 !(import! &self (library OmegaClaw-Core ./src/providers))
 !(import! &self (library OmegaClaw-Core ./profile/policy))
 !(import! &self (library OmegaClaw-Core ./src/fileio.py))
+!(import! &self (library OmegaClaw-Core ./src/action_risk.py))
 !(import! &self (library OmegaClaw-Core ./src/skills))
 !(import! &self (library OmegaClaw-Core ./src/websearch.py))
 !(import! &self (library OmegaClaw-Core ./src/memory))

src/action_risk.py

@@ -0,0 +1,77 @@
+"""Deterministic advisory risk classification for intended OmegaClaw actions.
+
+This module does not execute, inspect, or intercept an action.  It classifies
+only the skill name and argument supplied by the caller and never includes the
+argument in its result or logs.
+"""
+
+import re
+
+
+READ_ONLY_SKILLS = {
+    "action-risk-check",
+    "episodes",
+    "get-io-policy",
+    "query",
+    "read-file",
+    "search",
+    "technical-analysis",
+    "tavily-search",
+    "version",
+    "websearch",
+}
+
+STATE_CHANGING_SKILLS = {
+    "append-file",
+    "metta",
+    "pin",
+    "remember",
+    "send",
+    "shell",
+    "write-file",
+    "write-file-b64",
+}
+
+DESTRUCTIVE_PATTERNS = (
+    re.compile(r"(?:^|[;&|]\s*|\bsudo\s+)rm\s+(?:[^;&|]*\s)?-(?:[a-z]*r[a-z]*f|[a-z]*f[a-z]*r)\b", re.I),
+    re.compile(r"\b(?:docker\s+)?volume\s+(?:rm|remove|prune)\b", re.I),
+    re.compile(r"\bdocker\s+(?:system|container)\s+prune\b", re.I),
+    re.compile(r"\b(?:drop|truncate)\s+(?:table|database|schema)\b", re.I),
+    re.compile(r"\bgit\s+push\b[^\n;&|]*(?:--force(?:-with-lease)?\b|-f(?:\s|$))", re.I),
+    re.compile(r"\b(?:delete|destroy|erase|purge)\b[^\n]*(?:persistent\s+data|database|docker\s+volume)\b", re.I),
+)
+
+SENSITIVE_PATTERNS = (
+    re.compile(r"(?:^|[/\\])\.env(?:\.[^/\\\s]+)?(?:$|[/\\\s])", re.I),
+    re.compile(r"(?:^|[/\\])\.ssh(?:$|[/\\\s])", re.I),
+    re.compile(r"(?:^|[/\\])id_(?:rsa|dsa|ecdsa|ed25519)(?:\.pub)?(?:$|[/\\\s])", re.I),
+    re.compile(r"(?:^|[/\\])(?:credentials?|secrets?)(?:\.[^/\\\s]+)?(?:$|[/\\\s])", re.I),
+    re.compile(r"(?:^|[/\\])(?:shadow|passwd)(?:$|[/\\\s])", re.I),
+    re.compile(r"\b(?:telegram|openrouter)[_-]?(?:token|api[_-]?key)\b", re.I),
+    re.compile(r"\b(?:api[_-]?key|access[_-]?token|private[_-]?key)\s*[=:]", re.I),
+)
+
+
+def _normalize(value):
+    return " ".join(str(value).strip().lower().split())
+
+
+def _matches(patterns, text):
+    return any(pattern.search(text) for pattern in patterns)
+
+
+def check(skill_name, target_or_argument):
+    """Return a stable advisory classification without echoing caller input."""
+    skill = _normalize(skill_name).replace("_", "-")
+    target = _normalize(target_or_argument)
+
+    # Ordered deliberately: a lower-risk category must never weaken a match.
+    if _matches(DESTRUCTIVE_PATTERNS, target):
+        return "decision=block, risk=critical, reason=destructive-action"
+    if _matches(SENSITIVE_PATTERNS, target):
+        return "decision=review, risk=high, reason=sensitive-target"
+    if skill in STATE_CHANGING_SKILLS:
+        return "decision=review, risk=medium, reason=state-changing"
+    if skill in READ_ONLY_SKILLS:
+        return "decision=allow, risk=low, reason=read-only"
+    return "decision=review, risk=medium, reason=unknown-action"

src/helper.py

@@ -16,6 +16,7 @@
 
 TS_RE = re.compile(r'^\("(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})"')
 LLM_COMMANDS = {
+    "action-risk-check",
     "append-file",
     "episodes",
     "metta",
@@ -34,6 +35,7 @@
     "write-file-b64",
 }
 TWO_ARG_COMMANDS = {
+    "action-risk-check",
     "write-file",
     "append-file",
     "write-file-b64"
@@ -246,6 +248,9 @@ def test_omegaclaw_version():
 
 
 def test_balance_parenthesis():
+    assert balance_parentheses(
+        'action-risk-check read-file public/readme.txt\nversion'
+    ) == '((action-risk-check "read-file" "public/readme.txt") (version))'
     assert balance_parentheses('(write-file test.txt hello world)') == '((write-file "test.txt" "hello world"))'
     assert balance_parentheses('(append-file test.txt hello world)') == '((append-file "test.txt" "hello world"))'
     assert balance_parentheses('(write-file-b64 test.txt aGVsbG8=)') == '((write-file-b64 "test.txt" "aGVsbG8="))'

src/skills.metta

@@ -18,6 +18,7 @@
     "- Write base64-encoded content to file as a single line, prefer it when the content contains quotes, backslashes or multiple lines: write-file-b64 filename base64string"
     "- Append line to existing file, result read back from disk like write-file: append-file filename string"
     "- Get a list of allowed base paths for reading, writing, and updating files: get-io-policy"
+    "- Advisory deterministic risk classification for an intended action; this does not execute or automatically intercept the action: action-risk-check skill_name target_or_argument"
     ;COMMUNICATION CHANNELS:
     "- Send message to user: send string"
     "- Search the web: websearch string"
@@ -143,6 +144,9 @@
 (= (get-io-policy)
    (py-call (policy.get_allowed_policy_paths (securityPolicyPath))))
 
+(= (action-risk-check $skill $target)
+   (py-call (action_risk.check $skill $target)))
+
 !(import_prolog_functions_from_file (library OmegaClaw-Core ./src/skills.pl) (shell first_char gc read_file_tail))
 
 (= (metta $str)

tests/src_skills.metta

@@ -1,5 +1,6 @@
 !(import! &self ./tests/lib/utils)
 !(import! &self ./src/utils)
+!(import! &self ../src/action_risk.py)
 !(import! &self ./src/skills)
 !(import! &self ./src/loop)
 
@@ -13,6 +14,21 @@
          (expression-count-item (getSkills) "- This is a test skill: test-skill test_arg"))
        0)
 
+!(test (expression-count-item
+         (getSkills)
+         "- Advisory deterministic risk classification for an intended action; this does not execute or automatically intercept the action: action-risk-check skill_name target_or_argument")
+       1)
+
+!(test (contains-text
+         (action-risk-check "read-file" "public/readme.txt")
+         "decision=allow, risk=low, reason=read-only")
+       True)
+
+!(test (contains-text
+         (action-risk-check "shell" "docker volume rm omegaclaw-memory")
+         "decision=block, risk=critical, reason=destructive-action")
+       True)
+
 !(test (progn
          (add-prompt-extension test-prompt "TEST PROMPT EXTENSION")
          (contains-text (getPromptExtensions) "TEST PROMPT EXTENSION"))

tests/test_action_risk.py

@@ -0,0 +1,46 @@
+import unittest
+
+from src.action_risk import check
+
+
+class ActionRiskTests(unittest.TestCase):
+    def test_action_risk_classification(self):
+        cases = [
+            ("read-file", "public/readme.txt", "decision=allow, risk=low, reason=read-only"),
+            ("websearch", "read public password policy", "decision=allow, risk=low, reason=read-only"),
+            ("  READ-FILE  ", "  PUBLIC PASSWORD POLICY  ", "decision=allow, risk=low, reason=read-only"),
+            ("write-file", "notes.txt", "decision=review, risk=medium, reason=state-changing"),
+            ("unknown-skill", "ordinary argument", "decision=review, risk=medium, reason=unknown-action"),
+            ("read-file", "/project/.env", "decision=review, risk=high, reason=sensitive-target"),
+            ("read-file", "~/.ssh/id_ed25519", "decision=review, risk=high, reason=sensitive-target"),
+            ("read-file", "/home/user/credentials.json", "decision=review, risk=high, reason=sensitive-target"),
+            ("shell", "rm -rf /var/lib/app", "decision=block, risk=critical, reason=destructive-action"),
+            ("shell", "cd /srv && sudo rm -fr data", "decision=block, risk=critical, reason=destructive-action"),
+            ("shell", "docker volume rm omegaclaw-memory", "decision=block, risk=critical, reason=destructive-action"),
+            ("shell", "DROP TABLE users", "decision=block, risk=critical, reason=destructive-action"),
+            ("shell", "truncate database archive", "decision=block, risk=critical, reason=destructive-action"),
+            ("shell", "git push origin main --force", "decision=block, risk=critical, reason=destructive-action"),
+            ("shell", "git push -f origin main", "decision=block, risk=critical, reason=destructive-action"),
+        ]
+        for skill, target, expected in cases:
+            with self.subTest(skill=skill, target=target):
+                self.assertEqual(check(skill, target), expected)
+
+    def test_destructive_precedes_sensitive_and_state_changing(self):
+        self.assertEqual(
+            check("write-file", "rm -rf /project/.env"),
+            "decision=block, risk=critical, reason=destructive-action",
+        )
+
+    def test_sensitive_precedes_state_changing_and_read_only(self):
+        expected = "decision=review, risk=high, reason=sensitive-target"
+        self.assertEqual(check("write-file", "/project/.env"), expected)
+        self.assertEqual(check("read-file", "/project/.env"), expected)
+
+    def test_result_never_echoes_target(self):
+        marker = "DO-NOT-ECHO-THIS-VALUE"
+        self.assertNotIn(marker, check("read-file", f"credentials.json {marker}"))
+
+
+if __name__ == "__main__":
+    unittest.main()