feat(lattice): deprecate v1 ReasoningAuditor to advisory-only — move destructive patterns to Capability Lattice (closes #665)

agentwatch/core/recursion_depth_detector.py

@@ -40,10 +40,7 @@ def __init__(
         patterns: list[str] | None = None,
     ):
         self.threshold = threshold
-        self.patterns = [
-            re.compile(p, re.IGNORECASE)
-            for p in (patterns or DEFAULT_PATTERNS)
-        ]
+        self.patterns = [re.compile(p, re.IGNORECASE) for p in (patterns or DEFAULT_PATTERNS)]
         self.buffer = deque(maxlen=buffer_size)
 
     def observe(self, event: AgentEvent) -> RecursionDepthReport:
@@ -78,4 +75,4 @@ def reset(self):
 __all__ = [
     "RecursionDepthDetector",
     "RecursionDepthReport",
-]
+]

agentwatch/lattice/__init__.py

@@ -2,6 +2,16 @@
 
 from __future__ import annotations
 
+from agentwatch.lattice.attention_scatter import (
+    AttentionScatterDetector,
+    AttentionScatterReport,
+    compute_scatter_score,
+)
+from agentwatch.lattice.capability import (
+    CapabilitySignal,
+    CapabilitySignals,
+    detect_capability_signals,
+)
 from agentwatch.lattice.shadow_filesystem import (
     CRITICAL_SYSTEM_PATHS,
     FileAction,
@@ -12,10 +22,16 @@
 )
 
 __all__ = [
+    "AttentionScatterDetector",
+    "AttentionScatterReport",
     "CRITICAL_SYSTEM_PATHS",
+    "CapabilitySignal",
+    "CapabilitySignals",
     "FileAction",
     "FileOperation",
     "MutationResult",
     "MutationType",
     "ShadowFilesystem",
+    "compute_scatter_score",
+    "detect_capability_signals",
 ]

agentwatch/lattice/attention_scatter.py

@@ -0,0 +1,198 @@
+"""Cognitive Lattice — Attention Scatter Detector.
+
+Detects when an agent's attention scatters across unrelated paths or network domains,
+indicating loss of focus or potential rogue behaviour.
+
+Uses a sliding window of tool-call steps and a configurable entropy threshold.
+"""
+
+from __future__ import annotations
+
+import math
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+
+from agentwatch.core.schema import AgentEvent, EventType
+
+DEFAULT_WINDOW_SIZE = 20
+DEFAULT_SCATTER_THRESHOLD = 0.70
+
+
+def _safe_entropy(items: list[str]) -> float:
+    """Compute Shannon entropy for a list of discrete items.
+
+    Returns 0.0 for empty or single-element lists. For N discrete items, compute the
+    Shannon entropy and normalise it by the maximum entropy log2(n) to keep output
+    within the range [0.0, 1.0].
+    """
+    n = len(items)
+    if n <= 1:
+        return 0.0
+
+    frequencies: dict[str, int] = defaultdict(int)
+    for item in items:
+        frequencies[item] += 1
+
+    entropy = 0.0
+    for count in frequencies.values():
+        p = count / n
+        entropy -= p * math.log2(p)
+
+    max_entropy = math.log2(n) if n > 1 else 1.0
+    return entropy / max_entropy if max_entropy > 0 else 0.0
+
+
+def _extract_parent_dir(path: str) -> str:
+    """Return the parent directory of a path, or the path itself if it's a root."""
+    parent = path.rsplit("/", 1)[0]
+    return parent if parent else "/"
+
+
+def _domain_from_url(url: str) -> str:
+    """Extract domain from a URL. Returns the full string if unparseable."""
+    try:
+        parts = url.split("://")[1]
+        return parts.split("/")[0]
+    except (IndexError, AttributeError):
+        return url
+
+
+@dataclass
+class AttentionScatterReport:
+    """Result of a scatter check for a tool action.
+
+    Attributes:
+        blocked: Whether the scatter metric exceeds the safe threshold.
+        score: `[0.0, 1.0]` scatter metric (0 = focused, 1 = extreme scatter).
+        path_count: Number of unique paths in the window.
+        path_entropy: Shannon entropy of the parent directories in the window.
+        domain_count: Number of unique domains in the window.
+        domain_entropy: Shannon entropy of the domains in the window.
+        explanation: Human-readable summary for debugging.
+    """
+
+    blocked: bool
+    score: float
+    path_count: int
+    path_entropy: float
+    domain_count: int
+    domain_entropy: float
+    explanation: str
+
+    detected: bool = False
+    matches: int = 0
+    matched_patterns: list[str] = field(default_factory=list)
+
+
+@dataclass
+class AttentionScatterDetector:
+    """Detect attention scatter from tool-call traces.
+
+    Source:
+        V2 Cognitive Lattice — observe tool calls and verify the agent is not
+        inspecting unrelated files or domains across a region of steps.
+    """
+
+    window_size: int = DEFAULT_WINDOW_SIZE
+    scatter_threshold: float = DEFAULT_SCATTER_THRESHOLD
+
+    _path_window: deque[str] = field(default_factory=deque)
+    _domain_window: deque[str] = field(default_factory=deque)
+
+    def observe(self, event: AgentEvent) -> AttentionScatterReport:
+        """Feed one event into the scatter window and return a report."""
+        if event.event_type is not EventType.TOOL_CALL or event.tool_call is None:
+            return self._empty_safe_report()
+
+        tc = event.tool_call
+        affected = tc.affected_resources or []
+
+        paths: list[str] = []
+        domains: list[str] = []
+        for resource in affected:
+            if resource.startswith(("http://", "https://")):
+                domains.append(_domain_from_url(resource))
+            else:
+                paths.append(resource)
+
+        for p in paths:
+            self._path_window.append(p)
+        for d in domains:
+            self._domain_window.append(d)
+
+        return self._compute()
+
+    def _compute(self) -> AttentionScatterReport:
+        self._maintain_window_size()
+        path_list = list(self._path_window)
+        domain_list = list(self._domain_window)
+        path_ent = _safe_entropy(path_list) if path_list else 0.0
+        dom_ent = _safe_entropy(domain_list) if domain_list else 0.0
+        score = compute_scatter_score(
+            self._compute_path_entropy(),
+            self._compute_domain_entropy(),
+        )
+        blocked = score > self.scatter_threshold
+
+        explanation: list[str] = []
+        if len(path_list) >= 3:
+            explanation.append(
+                f"{len(path_list)} paths across {len(set(path_list))} directories (entropy {path_ent:.2f})"
+            )
+        if len(domain_list) >= 3:
+            explanation.append(f"{len(domain_list)} domains (entropy {dom_ent:.2f})")
+        if blocked:
+            explanation.insert(0, "BLOCKED: scatter exceeds threshold")
+        elif score >= self.scatter_threshold * 0.5:
+            explanation.insert(0, "WARN: scatter is elevated")
+
+        return AttentionScatterReport(
+            blocked=blocked,
+            matches=len(path_list) + len(domain_list),
+            matched_patterns=path_list + domain_list,
+            score=score,
+            path_count=len(set(path_list)),
+            path_entropy=path_ent,
+            domain_count=len(set(domain_list)),
+            domain_entropy=dom_ent,
+            explanation=" | ".join(explanation) if explanation else "",
+        )
+
+    def reset(self):
+        self._path_window.clear()
+        self._domain_window.clear()
+
+    def _empty_safe_report(self) -> AttentionScatterReport:
+        return AttentionScatterReport(
+            blocked=False,
+            score=0.0,
+            path_count=0,
+            path_entropy=0.0,
+            domain_count=0,
+            domain_entropy=0.0,
+            explanation="",
+        )
+
+    def _maintain_window_size(self) -> None:
+        while len(self._path_window) > self.window_size:
+            self._path_window.popleft()
+        while len(self._domain_window) > self.window_size:
+            self._domain_window.popleft()
+
+    def _compute_path_entropy(self) -> float:
+        return _safe_entropy([_extract_parent_dir(p) for p in self._path_window])
+
+    def _compute_domain_entropy(self) -> float:
+        return _safe_entropy(list(self._domain_window))
+
+
+def compute_scatter_score(path_entropy: float, domain_entropy: float) -> float:
+    """Compute combined scatter score (0.0..1.0)."""
+    return max(path_entropy, domain_entropy)
+
+
+__all__ = [
+    "AttentionScatterDetector",
+    "AttentionScatterReport",
+    "compute_scatter_score",
+]

agentwatch/lattice/capability.py

@@ -0,0 +1,141 @@
+"""v2 Capability Lattice — surface structural risk signals on tool calls.
+
+The Capability Lattice owns the legacy ``_DESTRUCTIVE_PATTERNS`` substring
+checks that used to live in ``agentwatch.reasoning.auditor``. Moving them
+out of the auditor makes the v2 architecture's separation of concerns
+explicit: the auditor is advisory-only; the Capability Lattice surfaces
+*structural* signals that the SafetyEngine (or a future v2 Safety Lattice)
+can use to block actions.
+
+This module is a pure function library — no I/O, no class state. The set
+of matched patterns is returned as a structured list that downstream
+consumers can render, log, or escalate.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from agentwatch.core.schema import AgentEvent, ToolCallData
+
+# Substrings that mark a shell/tool command as destructive. Kept intentionally
+# conservative and lowercase; matched against the command + string arguments.
+# Moved verbatim from ``agentwatch.reasoning.auditor._DESTRUCTIVE_PATTERNS``
+# when the v1 auditor was deprecated to advisory-only.
+_DESTRUCTIVE_PATTERNS: tuple[str, ...] = (
+    "rm -rf",
+    "rm -r ",
+    "rmdir",
+    "drop table",
+    "drop database",
+    "truncate ",
+    "mkfs",
+    "dd if=",
+    "shutdown",
+    "reboot",
+    "> /dev/sd",
+    ":(){:|:&};:",
+)
+
+# Mirrors the signal taxonomy that the v1 auditor emitted so downstream
+# dashboards keep working without code changes.
+PRIVILEGE_ESCALATION_MARKERS = ("sudo",)
+EXTERNAL_FETCH_MARKERS = ("curl ", "wget ", "http://", "https://")
+
+__all__ = [
+    "CapabilitySignal",
+    "CapabilitySignals",
+    "detect_capability_signals",
+]
+
+
+@dataclass(frozen=True)
+class CapabilitySignal:
+    """One structural signal surfaced by the Capability Lattice.
+
+    Attributes:
+        kind: Signal name (e.g. ``"destructive_command"``). Stable across versions
+            so dashboards and audit logs can match on it.
+        matched: The substring that triggered the signal. Useful for log-only
+            contexts where the actual pattern matched matters for forensics.
+    """
+
+    kind: str
+    matched: str
+
+
+@dataclass(frozen=True)
+class CapabilitySignals:
+    """Aggregate signal report for one tool call."""
+
+    signals: tuple[CapabilitySignal, ...]
+    has_destructive: bool
+    has_privilege_escalation: bool
+    has_broad_wildcard: bool
+    has_external_fetch: bool
+
+    def to_risk_signal_names(self) -> list[str]:
+        """Return the v1 risk-signal names so dashboards keep working."""
+        names: list[str] = []
+        if self.has_destructive:
+            names.append("destructive_command")
+        if self.has_privilege_escalation:
+            names.append("privilege_escalation")
+        if self.has_broad_wildcard:
+            names.append("broad_wildcard")
+        if self.has_external_fetch:
+            names.append("external_fetch")
+        return names
+
+
+def _command_text(tool_call: ToolCallData) -> str:
+    """Lowercased concatenation of ``raw_command`` and string arguments."""
+    command = (tool_call.raw_command or "").lower()
+    arg_text = (
+        " ".join(
+            str(value).lower() for value in tool_call.arguments.values() if isinstance(value, str)
+        )
+        if tool_call.arguments
+        else ""
+    )
+    return f"{command} {arg_text}".strip()
+
+
+def detect_capability_signals(event: AgentEvent) -> CapabilitySignals:
+    """Surface structural risk signals on a tool call.
+
+    Returns an empty :class:`CapabilitySignals` when the event has no
+    ``tool_call`` payload, or when no patterns matched.
+    """
+    tool_call = event.tool_call
+    if tool_call is None:
+        return CapabilitySignals((), False, False, False, False)
+
+    haystack = _command_text(tool_call)
+    signals: list[CapabilitySignal] = []
+
+    for pattern in _DESTRUCTIVE_PATTERNS:
+        if pattern in haystack:
+            signals.append(CapabilitySignal("destructive_command", pattern))
+            break
+
+    for marker in PRIVILEGE_ESCALATION_MARKERS:
+        if haystack.startswith(marker) or f"{marker} " in haystack:
+            signals.append(CapabilitySignal("privilege_escalation", marker))
+            break
+
+    if "*" in haystack:
+        signals.append(CapabilitySignal("broad_wildcard", "*"))
+
+    for marker in EXTERNAL_FETCH_MARKERS:
+        if marker in haystack:
+            signals.append(CapabilitySignal("external_fetch", marker))
+            break
+
+    return CapabilitySignals(
+        signals=tuple(signals),
+        has_destructive=any(s.kind == "destructive_command" for s in signals),
+        has_privilege_escalation=any(s.kind == "privilege_escalation" for s in signals),
+        has_broad_wildcard=any(s.kind == "broad_wildcard" for s in signals),
+        has_external_fetch=any(s.kind == "external_fetch" for s in signals),
+    )