From aa40b593a268511127bb2f0187178879d55440ff Mon Sep 17 00:00:00 2001 From: Archit Adish Gupta Date: Tue, 4 Aug 2026 20:59:17 +0530 Subject: [PATCH 1/2] =?UTF-8?q?feat(lattice):=20deprecate=20v1=20Reasoning?= =?UTF-8?q?Auditor=20to=20advisory-only=20=E2=80=94=20move=20destructive?= =?UTF-8?q?=20patterns=20to=20Capability=20Lattice=20(closes=20#665)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agentwatch/lattice/__init__.py | 16 ++ agentwatch/lattice/attention_scatter.py | 198 +++++++++++++++++ agentwatch/lattice/capability.py | 141 +++++++++++++ agentwatch/reasoning/auditor.py | 103 +++++---- tests/test_attention_scatter.py | 269 ++++++++++++++++++++++++ tests/test_auditor_advisory.py | 262 +++++++++++++++++++++++ 6 files changed, 945 insertions(+), 44 deletions(-) create mode 100644 agentwatch/lattice/attention_scatter.py create mode 100644 agentwatch/lattice/capability.py create mode 100644 tests/test_attention_scatter.py create mode 100644 tests/test_auditor_advisory.py diff --git a/agentwatch/lattice/__init__.py b/agentwatch/lattice/__init__.py index 8f6cde85..c9f15a58 100644 --- a/agentwatch/lattice/__init__.py +++ b/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", ] diff --git a/agentwatch/lattice/attention_scatter.py b/agentwatch/lattice/attention_scatter.py new file mode 100644 index 00000000..73bc4928 --- /dev/null +++ b/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", +] diff --git a/agentwatch/lattice/capability.py b/agentwatch/lattice/capability.py new file mode 100644 index 00000000..4b1feab6 --- /dev/null +++ b/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), + ) diff --git a/agentwatch/reasoning/auditor.py b/agentwatch/reasoning/auditor.py index 64d6c556..56cdf81a 100644 --- a/agentwatch/reasoning/auditor.py +++ b/agentwatch/reasoning/auditor.py @@ -1,4 +1,17 @@ -"""Reasoning step auditor with optional LLM-judge callback.""" +"""Reasoning step auditor with optional LLM-judge callback. + +.. deprecated:: + The :class:`ReasoningAuditor` is **advisory-only** as of v2 (Phase 5 — see + issue #665). It no longer participates in blocking decisions: low scores + produce a ``WARNING`` verdict but never trigger ``ExecutionStatus.BLOCKED``. + Blocking is now owned by the Capability Lattice (``agentwatch.lattice.capability``) + and the State Lattice (``agentwatch.lattice.shadow_filesystem``). + + The legacy ``_DESTRUCTIVE_PATTERNS`` substring matches that used to live + here have been moved to :mod:`agentwatch.lattice.capability` — the + auditor now imports them from there so dashboards reading + ``audit.risk_signals`` keep working unchanged. +""" from __future__ import annotations @@ -14,6 +27,7 @@ ReasoningStyleFingerprint, StyleSwapAlert, ) +from agentwatch.lattice.capability import detect_capability_signals from agentwatch.reasoning.fingerprint import ( StyleFingerprint, detect_mid_session_change, @@ -22,22 +36,21 @@ JudgeCallback = Callable[[str, AgentEvent], Awaitable[dict[str, Any]]] -# Substrings that mark a shell/tool command as destructive. Kept intentionally -# conservative and lowercase; matched against the command + string arguments. -_DESTRUCTIVE_PATTERNS = ( - "rm -rf", - "rm -r ", - "rmdir", - "drop table", - "drop database", - "truncate ", - "mkfs", - "dd if=", - "shutdown", - "reboot", - "> /dev/sd", - ":(){:|:&};:", -) +# Advisory verdict taxonomy (Phase 5 — see issue #665). +# Replaces the legacy v1 "sound"/"acceptable"/"weak" verdicts so consumers can +# distinguish an advisory warning from a block signal at a glance. +ADVISORY_CLEAN = "advisory_clean" +ADVISORY_NOTE = "advisory_note" +ADVISORY_WARN = "advisory_warn" + +# Back-compat aliases for any downstream consumer that still queries the +# pre-v2 verdict names. New code should use the ``advisory_*`` constants above. +LEGACY_VERDICT_SOUND = "sound" +LEGACY_VERDICT_ACCEPTABLE = "acceptable" +LEGACY_VERDICT_WEAK = "weak" + +ADVISORY_WARN_THRESHOLD = 0.40 +ADVISORY_NOTE_THRESHOLD = 0.70 @dataclass @@ -53,6 +66,9 @@ class StepAudit: risk_signals: list[str] = field(default_factory=list) confidence_breakdown: dict[str, float] = field(default_factory=dict) latency_ms: float = 0.0 + # Phase 5 (#665): make the advisory-only semantics explicit. New code can + # ignore this field; legacy consumers can rely on the v1 risk_signals list. + is_advisory_only: bool = True def to_dict(self) -> dict[str, object]: return { @@ -68,6 +84,7 @@ def to_dict(self) -> dict[str, object]: key: round(value, 3) for key, value in self.confidence_breakdown.items() }, "latency_ms": round(self.latency_ms, 2), + "is_advisory_only": self.is_advisory_only, } @@ -172,30 +189,19 @@ async def audit_step(self, step_index: int, event: AgentEvent) -> StepAudit: @staticmethod def detect_risk_signals(event: AgentEvent) -> list[str]: - """Surface human-readable risk signals for a step, e.g. destructive - commands or overly-broad wildcards, so a block is explainable.""" - signals: list[str] = [] - tool_call = event.tool_call - if tool_call is not None: - 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 "" - ) - haystack = f"{command} {arg_text}".strip() - if any(pattern in haystack for pattern in _DESTRUCTIVE_PATTERNS): - signals.append("destructive_command") - if haystack.startswith("sudo") or "sudo " in haystack: - signals.append("privilege_escalation") - if "*" in haystack: - signals.append("broad_wildcard") - if any(token in haystack for token in ("curl ", "wget ", "http://", "https://")): - signals.append("external_fetch") + """Surface human-readable risk signals for a step. + + Phase 5 (#665): the substring pattern matching now lives in the + Capability Lattice (``agentwatch.lattice.capability``). This method + stays as the canonical "v1 risk_signal name" surface so dashboards + and audit logs keep working unchanged. New code that needs to take + a blocking decision should call :func:`detect_capability_signals` + directly. + + Tool-result-error and ``blocked_action`` markers remain here because + they describe the *outcome* of an action, not its structural risk. + """ + signals = detect_capability_signals(event).to_risk_signal_names() if event.tool_result is not None and event.tool_result.error: signals.append("tool_error") if event.is_blocked: @@ -321,10 +327,19 @@ def _heuristic_audit(self, step_index: int, event: AgentEvent) -> StepAudit: evidence.append("blocked_action") score = max(0.0, min(score, 1.0)) - verdict = "sound" if score >= 0.75 else "acceptable" if score >= 0.5 else "weak" + # Phase 5 (#665): advisory-only verdict taxonomy. Replaces the v1 + # "sound"/"acceptable"/"weak" names so consumers can tell at a glance + # that this output does not trigger any blocking decision. + if score >= ADVISORY_NOTE_THRESHOLD: + verdict = ADVISORY_CLEAN + elif score >= ADVISORY_WARN_THRESHOLD: + verdict = ADVISORY_NOTE + else: + verdict = ADVISORY_WARN rationale = ( - "Heuristic audit based on observability artifacts because no external judge " - "callback is configured." + "Advisory heuristic audit based on observability artifacts because no " + "external judge callback is configured. This verdict is advisory-only — " + "blocking decisions are owned by the Capability and State lattices." ) # Interpretable per-dimension decomposition (does not alter `score`). diff --git a/tests/test_attention_scatter.py b/tests/test_attention_scatter.py new file mode 100644 index 00000000..7d548167 --- /dev/null +++ b/tests/test_attention_scatter.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +from agentwatch.core.schema import AgentEvent, EventType, ToolCallData +from agentwatch.lattice.attention_scatter import ( + AttentionScatterDetector, + _domain_from_url, + _extract_parent_dir, + _safe_entropy, + compute_scatter_score, +) + + +def _event( + step: int, + affected_resources: list[str] | None = None, + tool_name: str = "read_file", +) -> AgentEvent: + return AgentEvent( + event_type=EventType.TOOL_CALL, + session_id="s1", + agent_id="a1", + step_number=step, + timestamp=datetime.now(UTC), + tool_call=ToolCallData( + tool_name=tool_name, + affected_resources=affected_resources or [], + ), + ) + + +# ----------------------------------------------------------------- entropy helper + + +def test_entropy_empty(): + assert _safe_entropy([]) == 0.0 + + +def test_entropy_single(): + assert _safe_entropy(["a"]) == 0.0 + + +def test_entropy_uniform(): + assert _safe_entropy(["a", "b", "c"]) > 0.95 + + +def test_entropy_skewed(): + assert _safe_entropy(["a", "a", "a", "b"]) < 0.9 + + +# ----------------------------------------------------------------- parent dir + + +def test_parent_dir_deep(): + assert _extract_parent_dir("/workspace/src/module/file.py") == "/workspace/src/module" + + +def test_parent_dir_shallow(): + assert _extract_parent_dir("/workspace/readme.md") == "/workspace" + + +# ----------------------------------------------------------------- domain + + +def test_domain_from_https(): + assert _domain_from_url("https://api.example.com/v1/data") == "api.example.com" + + +def test_domain_from_http(): + assert _domain_from_url("http://localhost:8080/status") == "localhost:8080" + + +def test_domain_unparseable(): + assert _domain_from_url("not-a-url") == "not-a-url" + + +# ----------------------------------------------------------------- scorer + + +def test_scorer_zero(): + assert compute_scatter_score(0.0, 0.0) == 0.0 + + +def test_scorer_path_dominates(): + assert compute_scatter_score(0.9, 0.2) == 0.9 + + +def test_scorer_domain_dominates(): + assert compute_scatter_score(0.1, 0.85) == 0.85 + + +# ----------------------------------------------------------------- detector — non-tool events pass through safely + + +def test_non_tool_event_ignored(): + det = AttentionScatterDetector() + result = det.observe( + AgentEvent( + event_type=EventType.SESSION_START, + session_id="s1", + agent_id="a1", + timestamp=datetime.now(UTC), + ) + ) + assert result.blocked is False + assert result.score == 0.0 + assert result.explanation == "" + + +# ----------------------------------------------------------------- focused access (same directory, same domain) + + +def test_focused_path_access_is_safe(): + det = AttentionScatterDetector(window_size=10, scatter_threshold=0.70) + for step in range(5): + result = det.observe(_event(step, ["/workspace/src/utils.py"])) + assert result.blocked is False, f"step {step}" + assert result.score == 0.0, f"step {step}: score={result.score}" + + +def test_focused_domain_access_is_safe(): + det = AttentionScatterDetector(window_size=10, scatter_threshold=0.70) + for step in range(4): + result = det.observe(_event(step, ["https://api.openai.com/v1/chat/completions"])) + assert result.blocked is False, f"step {step}" + + +# ----------------------------------------------------------------- moderate scatter (warn, not block) + + +def test_modest_scatter_warns(): + # Use threshold just above 1.0 so even a 3-distinct-paths window won't block. + det = AttentionScatterDetector(window_size=10, scatter_threshold=1.5) + # 3 distinct paths across 3 dirs → entropy 1.0 (max possible for N=3) — must NOT block + det.observe(_event(1, ["/workspace/src/main.py"])) + det.observe(_event(2, ["/workspace/tests/test.py"])) + result = det.observe(_event(3, ["/workspace/docs/readme.md"])) + assert result.blocked is False, f"expected no block at threshold 1.5, got {result.explanation}" + assert result.path_count == 3 + assert result.path_entropy > 0.0 + + +# ----------------------------------------------------------------- scatter BLOCKED (many distinct directories) + + +def test_wide_scatter_is_blocked(): + det = AttentionScatterDetector(window_size=10, scatter_threshold=0.50) + dirs = [ + "/etc/passwd", + "/home/user/.bashrc", + "/var/log/syslog", + "/tmp/session", # noqa: S108 (test fixture, not a real path) + "/boot/efi", + "/opt/tool", + "/srv/backup", + ] + for i, path in enumerate(dirs): + result = det.observe(_event(i, [path])) + assert result.blocked is True + assert "BLOCKED" in result.explanation + assert result.score > 0.50 + + +# ----------------------------------------------------------------- many domains = blocked + + +def test_many_domains_blocked(): + det = AttentionScatterDetector(window_size=8, scatter_threshold=0.50) + domains = [ + "https://api.openai.com/chat", + "https://api.anthropic.com/messages", + "https://api.google.com/ai", + "https://api.mistral.ai/chat", + "https://api.cohere.ai/chat", + "https://api.groq.com/chat", + ] + for i, domain in enumerate(domains): + result = det.observe(_event(i, [domain])) + assert result.blocked is True + assert "BLOCKED" in result.explanation + + +# ----------------------------------------------------------------- reset + + +def test_reset_clears_state(): + det = AttentionScatterDetector(window_size=5, scatter_threshold=0.50) + for step in range(5): + det.observe(_event(step, [f"/workspace/module_{step}/file.py"])) + assert ( + det.observe( + _event(5, ["/etc/passwd"]), + ).blocked + is True + ) + det.reset() + assert ( + det.observe( + _event(6, ["/workspace/src/main.py"]), + ).blocked + is False + ) + + +# ----------------------------------------------------------------- window overflow + + +def test_window_overflow_drops_old_entries(): + det = AttentionScatterDetector(window_size=3, scatter_threshold=0.50) + det.observe(_event(1, ["/a/x.txt"])) + det.observe(_event(2, ["/b/y.txt"])) + det.observe(_event(3, ["/c/z.txt"])) + # window now contains 3 paths across 3 dirs → scatter + result = det.observe(_event(4, ["/a/q.txt"])) + # old "/a/x.txt" dropped out, "/c/z.txt" still present → should be blocked + assert result.blocked + # next 3 steps stabilize + det.observe(_event(5, ["/a/p.txt"])) + det.observe(_event(6, ["/a/o.txt"])) + result = det.observe(_event(7, ["/a/n.txt"])) + assert result.blocked is False, ( + f"expected safety after saturation, got blocked score={result.score}" + ) + + +# ----------------------------------------------------------------- scored quantiles + + +def test_score_0_for_identical_items(): + det = AttentionScatterDetector(window_size=4) + for i in range(4): + result = det.observe(_event(i, ["/src/main.py"])) + assert result.score == 0.0 + + +def test_score_increases_with_disparity(): + det = AttentionScatterDetector(window_size=4) + det.observe(_event(1, ["/src/main.py"])) + result2 = det.observe(_event(2, ["/src/utils.py"])) + assert result2.score == 0.0 + result3 = det.observe(_event(3, ["/tests/test.py"])) + assert result3.score > 0.0, result3.explanation + result4 = det.observe(_event(4, ["/docs/readme.md"])) + assert result4.score > result3.score + + +# ----------------------------------------------------------------- mixed resources + + +def test_paths_and_domains_mixed(): + det = AttentionScatterDetector(window_size=4, scatter_threshold=0.95) + det.observe(_event(1, ["https://api.openai.com/chat", "/workspace/src/main.py"])) + det.observe(_event(2, ["https://api.anthropic.com/messages", "/workspace/tests/test.py"])) + result = det.observe(_event(3, ["https://api.google.com/ai", "/workspace/docs/readme.md"])) + # With high threshold 0.95 and 6 distinct items, scatter IS at 1.0 → blocked + assert result.blocked is True, f"expected BLOCKED with high scatter, got {result.explanation}" + assert result.path_count == 3 + assert result.domain_count == 3 + + +# ----------------------------------------------------------------- reset on create + + +def test_fresh_detector_has_zero_entropy(): + det = AttentionScatterDetector() + result = det.observe(_event(1, ["/src/main.py"])) + assert result.score == 0.0 + assert result.blocked is False diff --git a/tests/test_auditor_advisory.py b/tests/test_auditor_advisory.py new file mode 100644 index 00000000..a2320af6 --- /dev/null +++ b/tests/test_auditor_advisory.py @@ -0,0 +1,262 @@ +"""Tests for the Phase 5 (#665) advisory-only refactor of ``ReasoningAuditor``. + +The v1 auditor emitted verdicts ``"sound" / "acceptable" / "weak"`` and lived +next to substring pattern matching. The v2 refactor promotes an explicit +advisory taxonomy (``advisory_clean`` / ``advisory_note`` / ``advisory_warn``) +and moves the legacy regex matching to the Capability Lattice. + +These tests pin the new surface while preserving the v1 risk-signal names +so dashboards keep working. +""" + +from __future__ import annotations + +import asyncio + +from agentwatch.core.schema import ( + AgentEvent, + AgentFramework, + EventType, + ExecutionStatus, + ToolCallData, +) +from agentwatch.lattice.capability import ( + CapabilitySignal, + CapabilitySignals, + detect_capability_signals, +) +from agentwatch.reasoning.auditor import ( + ADVISORY_CLEAN, + ADVISORY_NOTE, + ADVISORY_WARN, + LEGACY_VERDICT_ACCEPTABLE, + LEGACY_VERDICT_SOUND, + LEGACY_VERDICT_WEAK, + ReasoningAuditor, + StepAudit, +) + + +def _tool_call( + tool: str, + args: dict | None = None, + raw: str | None = None, + *, + status: ExecutionStatus = ExecutionStatus.RUNNING, +) -> AgentEvent: + return AgentEvent( + session_id="S", + agent_id="A", + framework=AgentFramework.CUSTOM, + event_type=EventType.TOOL_CALL, + status=status, + tool_call=ToolCallData(tool_name=tool, arguments=args or {}, raw_command=raw), + ) + + +# ----------------------------------------------------------------- verdict taxonomy + + +def test_verdict_constants_distinct(): + assert {ADVISORY_CLEAN, ADVISORY_NOTE, ADVISORY_WARN} == { + "advisory_clean", + "advisory_note", + "advisory_warn", + } + + +def test_legacy_verdict_aliases_preserved(): + # v1 dashboards may still query the old names — they must be importable. + assert LEGACY_VERDICT_SOUND == "sound" + assert LEGACY_VERDICT_ACCEPTABLE == "acceptable" + assert LEGACY_VERDICT_WEAK == "weak" + + +# ----------------------------------------------------------------- StepAudit advisory-only flag + + +def test_step_audit_is_advisory_only_default(): + audit = StepAudit(step_index=0, event_id="e", score=0.9, verdict="x", rationale="ok") + assert audit.is_advisory_only is True + + +def test_step_audit_to_dict_exposes_advisory_flag(): + audit = StepAudit(step_index=0, event_id="e", score=0.9, verdict="x", rationale="ok") + payload = audit.to_dict() + assert payload["is_advisory_only"] is True + + +# ----------------------------------------------------------------- heuristic verdicts + + +def test_heuristic_high_score_is_advisory_clean(): + auditor = ReasoningAuditor() + # A planner-only event with a long output gets score 0.85 (≥ 0.70 threshold). + event = AgentEvent( + session_id="S", + agent_id="A", + framework=AgentFramework.CUSTOM, + event_type=EventType.PLANNER_OUTPUT, + planner_output_preview="Step one then step two then step three then step four then step five then step six then step seven then step eight", + ) + audit = asyncio.run(auditor.audit_step(0, event)) + assert audit.verdict == ADVISORY_CLEAN, audit.to_dict() + + +def test_heuristic_low_score_is_advisory_warn(): + auditor = ReasoningAuditor() + # An event with no planner text, no tool args, and a tool_result error gets + # score 0.55 - 0.25 = 0.30 (under the 0.40 warn threshold). + from agentwatch.core.schema import ToolResultData + + event = AgentEvent( + session_id="S", + agent_id="A", + framework=AgentFramework.CUSTOM, + event_type=EventType.TOOL_CALL, + tool_call=ToolCallData(tool_name="stub"), + tool_result=ToolResultData( + tool_name="stub", + tool_id="t1", + output="", + error="command not found", + ), + ) + audit = asyncio.run(auditor.audit_step(0, event)) + assert audit.verdict == ADVISORY_WARN, audit.to_dict() + + +def test_heuristic_rationale_mentions_advisory_only(): + auditor = ReasoningAuditor() + event = _tool_call("shell", {"command": "ls"}, raw="ls") + audit = asyncio.run(auditor.audit_step(0, event)) + assert "advisory-only" in audit.rationale + assert "lattices" in audit.rationale + + +# ----------------------------------------------------------------- Capability Lattice + + +def test_detect_capability_signals_empty_event(): + event = AgentEvent( + session_id="S", + agent_id="A", + framework=AgentFramework.CUSTOM, + event_type=EventType.SESSION_START, + ) + result = detect_capability_signals(event) + assert result.signals == () + assert result.has_destructive is False + assert result.has_privilege_escalation is False + assert result.has_broad_wildcard is False + assert result.has_external_fetch is False + + +def test_detect_capability_signals_destructive(): + event = _tool_call("shell", {"command": "rm -rf /"}, raw="rm -rf /") + result = detect_capability_signals(event) + assert result.has_destructive is True + kinds = {s.kind for s in result.signals} + assert "destructive_command" in kinds + + +def test_detect_capability_signals_privilege_escalation(): + event = _tool_call("shell", {"command": "sudo apt-get update"}, raw="sudo apt-get update") + result = detect_capability_signals(event) + assert result.has_privilege_escalation is True + + +def test_detect_capability_signals_broad_wildcard(): + event = _tool_call("shell", {"command": "ls *"}, raw="ls *") + result = detect_capability_signals(event) + assert result.has_broad_wildcard is True + + +def test_detect_capability_signals_external_fetch(): + event = _tool_call( + "curl", {"command": "curl https://example.com"}, raw="curl https://example.com" + ) + result = detect_capability_signals(event) + assert result.has_external_fetch is True + + +def test_detect_capability_signals_to_risk_signal_names_round_trip(): + event = _tool_call( + "shell", + {"command": "sudo rm -rf /tmp/* && curl https://x"}, + raw="sudo rm -rf /tmp/* && curl https://x", + ) + result = detect_capability_signals(event) + names = result.to_risk_signal_names() + assert "destructive_command" in names + assert "privilege_escalation" in names + assert "broad_wildcard" in names + assert "external_fetch" in names + + +# ----------------------------------------------------------------- CapabilitySignals dataclass + + +def test_capability_signals_is_frozen(): + signals = CapabilitySignals((), False, False, False, False) + try: + signals.has_destructive = True # type: ignore[misc] + raise AssertionError("expected frozen dataclass to raise") + except AttributeError: + pass + + +def test_capability_signal_is_frozen(): + sig = CapabilitySignal("destructive_command", "rm -rf") + try: + sig.kind = "x" # type: ignore[misc] + raise AssertionError("expected frozen dataclass to raise") + except AttributeError: + pass + + +# ----------------------------------------------------------------- auditor delegates to lattice + + +def test_auditor_risk_signals_match_capability_lattice_names(): + auditor = ReasoningAuditor() + event = _tool_call( + "shell", + {"command": "sudo rm -rf /var/*"}, + raw="sudo rm -rf /var/*", + ) + audit = asyncio.run(auditor.audit_step(0, event)) + # The auditor's v1-style ``risk_signals`` list must match what the lattice would say. + expected = detect_capability_signals(event).to_risk_signal_names() + for name in expected: + assert name in audit.risk_signals, f"{name!r} missing from auditor's risk_signals" + + +def test_auditor_does_not_modify_event_status(): + """The auditor must never flip an event's status to BLOCKED.""" + auditor = ReasoningAuditor() + event = _tool_call( + "shell", + {"command": "rm -rf /"}, + raw="rm -rf /", + ) + asyncio.run(auditor.audit_step(0, event)) + assert event.status != ExecutionStatus.BLOCKED, ( + "Phase 5 (#665): auditor is advisory-only and must not set BLOCKED status" + ) + + +# ----------------------------------------------------------------- regression — existing risk_signals surface + + +def test_blocked_action_signal_emitted_when_event_is_blocked(): + """Legacy risk-signal surface must still flag pre-blocked events.""" + auditor = ReasoningAuditor() + event = _tool_call( + "shell", + {"command": "echo hi"}, + raw="echo hi", + status=ExecutionStatus.BLOCKED, + ) + audit = asyncio.run(auditor.audit_step(0, event)) + assert "blocked_action" in audit.risk_signals From d9b011c00c1fc57212ca58c38b8e4a64d34aa779 Mon Sep 17 00:00:00 2001 From: Archit Adish Gupta Date: Fri, 7 Aug 2026 22:31:22 +0530 Subject: [PATCH 2/2] style(core): ruff format recursion_depth_detector.py for ruff 0.16 --- agentwatch/core/recursion_depth_detector.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/agentwatch/core/recursion_depth_detector.py b/agentwatch/core/recursion_depth_detector.py index aff1eb6c..3b6f5d0d 100644 --- a/agentwatch/core/recursion_depth_detector.py +++ b/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", -] \ No newline at end of file +]