feat(lattice): deprecate v1 ReasoningAuditor to advisory-only — move destructive patterns to Capability Lattice (closes #665) - #688
Open
arcgod-design wants to merge 2 commits into
Conversation
…destructive patterns to Capability Lattice (closes sreerevanth#665)
Contributor
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughThe PR adds attention-scatter and Capability Lattice APIs. It changes ChangesLattice detection and advisory auditing
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant AgentEvent
participant ReasoningAuditor
participant CapabilityLattice
participant StepAudit
AgentEvent->>ReasoningAuditor: audit event
ReasoningAuditor->>CapabilityLattice: detect_capability_signals(event)
CapabilityLattice-->>ReasoningAuditor: return capability signals
ReasoningAuditor->>StepAudit: record advisory verdict and rationale
StepAudit-->>AgentEvent: preserve execution status
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
🧪 PR Test Results
Python 3.12 · commit d9b011c |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (1)
tests/test_attention_scatter.py (1)
132-141: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the warning result.
The test verifies only that the detector does not block. The test passes if the warning branch is removed. Assert the
"WARN:"prefix to verify the behavior named by the test.Proposed fix
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.explanation.startswith("WARN:") assert result.path_count == 3🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_attention_scatter.py` around lines 132 - 141, Update test_modest_scatter_warns to assert that result.explanation starts with the "WARN:" prefix, while preserving the existing non-blocking, path-count, and entropy assertions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@agentwatch/lattice/attention_scatter.py`:
- Around line 96-97: Validate window_size during initialization of the class
owning the window_size field, rejecting zero and negative values with an
appropriate exception before scatter processing begins. Preserve the existing
positive-window behavior and prevent the loop around the deque access from
receiving an empty window.
- Around line 127-158: Update _compute() to build the parent-directory list once
from the collected paths, then use that list consistently for path_entropy,
compute_scatter_score, and path_count. Replace the full-path list in those
metric calculations and ensure the explanation’s directory count reflects the
same parent-directory data, while preserving matched_patterns and path match
counts.
- Around line 99-100: Replace the separate _path_window and _domain_window
tracking in the attention-scatter state with a single window of paired
path/domain entries for each tool-call step. Update the append logic around the
tool-call handling and expire complete paired steps in the existing
window-maintenance logic, so path-only calls still remove prior domain accesses.
Add coverage for domain calls followed by more than window_size path-only calls
and verify the domain score no longer includes expired entries.
- Around line 51-57: Update _domain_from_url to use urlsplit for
case-insensitive HTTP/HTTPS URLs, returning the normalized hostname authority
with its port while excluding userinfo and query data, and preserving the
existing fallback for unparseable values. In the resource classification flow
around the HTTP/HTTPS check, reuse the same parsed URL result or normalized
domain logic so classification and domain-key generation process identical
values.
In `@agentwatch/reasoning/auditor.py`:
- Around line 330-342: Unify verdict classification through a shared
score-to-advisory-verdict helper, and use it for both heuristic and judge-backed
audits so StepAudit.verdict always emits advisory values when is_advisory_only
is true. Update the judge path around its existing verdict handling to retain
any judge-specific label separately if needed, while preserving the score-based
advisory result; add a regression test covering a configured judge returning a
non-advisory verdict.
In `@tests/test_auditor_advisory.py`:
- Around line 221-232: Update
test_auditor_risk_signals_match_capability_lattice_names to assert that
audit.risk_signals exactly equals
detect_capability_signals(event).to_risk_signal_names(), preserving the expected
ordering and rejecting duplicates or stale signals instead of checking
membership individually.
- Around line 243-246: Update the test around auditor.audit_step() to capture
event.status before invocation, then assert the status is equal to that captured
value afterward. Replace the current non-BLOCKED assertion while preserving the
existing advisory-only test context.
---
Nitpick comments:
In `@tests/test_attention_scatter.py`:
- Around line 132-141: Update test_modest_scatter_warns to assert that
result.explanation starts with the "WARN:" prefix, while preserving the existing
non-blocking, path-count, and entropy assertions.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 733cc6d8-6fd9-4772-87d0-c6347041b188
📒 Files selected for processing (6)
agentwatch/lattice/__init__.pyagentwatch/lattice/attention_scatter.pyagentwatch/lattice/capability.pyagentwatch/reasoning/auditor.pytests/test_attention_scatter.pytests/test_auditor_advisory.py
Comment on lines
+51
to
+57
| 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 |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the relevant module and usages.
fd -a attention_scatter.py . || true
if [ -f agentwatch/lattice/attention_scatter.py ]; then
wc -l agentwatch/lattice/attention_scatter.py
sed -n '1,160p' agentwatch/lattice/attention_scatter.py
fi
# Search domain entropy calculation for broader context.
rg -n "domain_entropy|domain_win|DOMAIN|_domain_from_url|startswith\\(\"https?\"|startswith\\(\"\\\"" agentwatch -S || trueRepository: sreerevanth/AgentWatch
Length of output: 7214
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Behavioral probe for current parsing and observed URL classification behavior.
python3 - <<'PY'
from urllib.parse import urlsplit
def _domain_from_url(url: str) -> str:
try:
parts = url.split("://")[1]
return parts.split("/")[0]
except (IndexError, AttributeError):
return url
def classify_current(resource: str):
if resource.startswith(("http://", "https://")):
return "domain", _domain_from_url(resource)
return "path", resource
cases = [
"https://api.example.com?request=1",
"https://api.example.com:443/api?request=1",
"https://user:pass@api.example.com/path",
"HTTP://api.example.com/path",
"HTTPS://api.example.com/path",
"file:///etc/hosts",
"ftps://example.com/data",
]
for c in cases:
kind, value = classify_current(c)
proposed = urlsplit(c)
proposed_value = (
proposed netloc.rsplit("@", 1)[-1].lower()
if proposed.scheme.lower() in {"http", "https"} and proposed.netloc
else c
)
print(f"{c!r}")
print(f" current: {kind}={value!r}")
print(f" urlsplit: scheme={proposed.scheme!r} netloc={proposed.netloc!r}")
print(f" proposed: proposed_value={proposed_value!r}")
PY
# Inspect tests around domain/path classification and domain entropy.
sed -n '1,240p' tests/test_attention_scatter.py
rg -n "_domain|domain_entropy|entropy|http://|https://|ATTEND|tool_call|affected" tests -S || trueRepository: sreerevanth/AgentWatch
Length of output: 276
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from urllib.parse import urlsplit
def _domain_from_url(url: str) -> str:
try:
parts = url.split("://")[1]
return parts.split("/")[0]
except (IndexError, AttributeError):
return url
def classify_current(resource: str):
if resource.startswith(("http://", "https://")):
return "domain", _domain_from_url(resource)
return "path", resource
cases = [
"https://api.example.com?request=1",
"https://api.example.com:443/api?request=1",
"https://user:pass@api.example.com/path",
"HTTP://api.example.com/path",
"HTTPS://api.example.com/path",
"file:///etc/hosts",
"ftps://example.com/data",
]
for c in cases:
current, value = classify_current(c)
parsed = urlsplit(c)
proposed = (
parsed.netloc.rsplit("@", 1)[-1].lower()
if parsed.scheme.lower() in {"http", "https"} and parsed.netloc
else c
)
print(f"{c!r}")
print(f" current: {current}={value!r}")
print(f" urlsplit: scheme={parsed.scheme!r} netloc={parsed.netloc!r}")
print(f" proposed: {proposed!r}")
PY
echo "--- tests.py domains/entropy/resource handling ---"
sed -n '1,280p' tests/test_attention_scatter.py || true
rg -n "_domain|domain_entropy|entropy|http://|https://|ATTEND|tool_call|affected|AttentionScatterDetector|scattered" tests -S || trueRepository: sreerevanth/AgentWatch
Length of output: 35698
Normalize URL authorities before using them as domain keys.
_domain_from_url() splits text, so https://api.example.com?request=1 becomes api.example.com?request=1 and mixed-case HTTP://api.example.com/path is classified as a path. Parse resources with urlsplit() instead, accept HTTP/HTTPS case-insensitively, keep the port but drop userinfo and query data from the domain key. Also apply the parser to the resource classification at lines 112-116 so the same value is used for both steps.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/lattice/attention_scatter.py` around lines 51 - 57, Update
_domain_from_url to use urlsplit for case-insensitive HTTP/HTTPS URLs, returning
the normalized hostname authority with its port while excluding userinfo and
query data, and preserving the existing fallback for unparseable values. In the
resource classification flow around the HTTP/HTTPS check, reuse the same parsed
URL result or normalized domain logic so classification and domain-key
generation process identical values.
Comment on lines
+96
to
+97
| window_size: int = DEFAULT_WINDOW_SIZE | ||
| scatter_threshold: float = DEFAULT_SCATTER_THRESHOLD |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Reject non-positive window_size values during initialization.
Line 96 permits window_size=-1. At Line 177, the loop continues after the deque becomes empty and raises IndexError. A zero size also silently disables scatter detection.
Proposed fix
class AttentionScatterDetector:
@@
_path_window: deque[str] = field(default_factory=deque)
_domain_window: deque[str] = field(default_factory=deque)
+ def __post_init__(self) -> None:
+ if self.window_size < 1:
+ raise ValueError("window_size must be at least 1")
+
def observe(self, event: AgentEvent) -> AttentionScatterReport:Also applies to: 176-180
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/lattice/attention_scatter.py` around lines 96 - 97, Validate
window_size during initialization of the class owning the window_size field,
rejecting zero and negative values with an appropriate exception before scatter
processing begins. Preserve the existing positive-window behavior and prevent
the loop around the deque access from receiving an empty window.
Comment on lines
+99
to
+100
| _path_window: deque[str] = field(default_factory=deque) | ||
| _domain_window: deque[str] = field(default_factory=deque) |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Track complete tool-call steps in one window.
Lines 118-121 append paths and domains to separate resource-count windows. Lines 176-180 expire entries only from the matching resource type. After more than window_size path-only calls, old domain accesses can remain in _domain_window and affect the next domain score.
Store paths and domains together for each tool-call step. Expire one complete step at a time. Add a test with domain calls followed by more than window_size path-only calls.
Proposed structure
- _path_window: deque[str] = field(default_factory=deque)
- _domain_window: deque[str] = field(default_factory=deque)
+ _event_window: deque[tuple[list[str], list[str]]] = field(default_factory=deque)
@@
- for p in paths:
- self._path_window.append(p)
- for d in domains:
- self._domain_window.append(d)
+ self._event_window.append((paths, domains))
@@
- path_list = list(self._path_window)
- domain_list = list(self._domain_window)
+ path_list = [path for paths, _ in self._event_window for path in paths]
+ domain_list = [domain for _, domains in self._event_window for domain in domains]
@@
- self._path_window.clear()
- self._domain_window.clear()
+ self._event_window.clear()
@@
- while len(self._path_window) > self.window_size:
- self._path_window.popleft()
- while len(self._domain_window) > self.window_size:
- self._domain_window.popleft()
+ while len(self._event_window) > self.window_size:
+ self._event_window.popleft()Also applies to: 118-123, 176-186
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/lattice/attention_scatter.py` around lines 99 - 100, Replace the
separate _path_window and _domain_window tracking in the attention-scatter state
with a single window of paired path/domain entries for each tool-call step.
Update the append logic around the tool-call handling and expire complete paired
steps in the existing window-maintenance logic, so path-only calls still remove
prior domain accesses. Add coverage for domain calls followed by more than
window_size path-only calls and verify the domain score no longer includes
expired entries.
Comment on lines
+127
to
+158
| 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 "", |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Report the same path metric used for blocking.
Line 129 calculates path_entropy from full paths. Lines 132 and 183 calculate the block score from parent directories. For /src/a.py and /src/b.py, the report can show path_entropy == 1.0 while score == 0.0. Line 140 also labels a full-path count as directories.
Build parent directories once in _compute(). Use that list for path_entropy, the score, and the directory count.
Proposed fix
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
+ parent_dirs = [_extract_parent_dir(path) for path in path_list]
+ path_ent = _safe_entropy(parent_dirs)
dom_ent = _safe_entropy(domain_list) if domain_list else 0.0
- score = compute_scatter_score(
- self._compute_path_entropy(),
- self._compute_domain_entropy(),
- )
+ score = compute_scatter_score(path_ent, dom_ent)
@@
explanation.append(
- f"{len(path_list)} paths across {len(set(path_list))} directories (entropy {path_ent:.2f})"
+ f"{len(path_list)} paths across {len(set(parent_dirs))} directories (entropy {path_ent:.2f})"
)Also applies to: 182-186
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/lattice/attention_scatter.py` around lines 127 - 158, Update
_compute() to build the parent-directory list once from the collected paths,
then use that list consistently for path_entropy, compute_scatter_score, and
path_count. Replace the full-path list in those metric calculations and ensure
the explanation’s directory count reflects the same parent-directory data, while
preserving matched_patterns and path match counts.
Comment on lines
+330
to
+342
| # 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." |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Emit advisory verdicts for judge-backed audits.
Line 333 through Line 338 classify only heuristic audits. Line 170 copies the judge verdict unchanged and defaults it to "uncertain". A configured judge can therefore emit "sound", "weak", or another non-advisory value while is_advisory_only remains True.
Use one score-to-advisory-verdict helper in both paths. Preserve a judge-specific label outside StepAudit.verdict if needed. Add a judge-callback regression test.
Proposed fix
+ `@staticmethod`
+ def _advisory_verdict(score: float) -> str:
+ if score >= ADVISORY_NOTE_THRESHOLD:
+ return ADVISORY_CLEAN
+ if score >= ADVISORY_WARN_THRESHOLD:
+ return ADVISORY_NOTE
+ return ADVISORY_WARN
+
async def audit_step(self, step_index: int, event: AgentEvent) -> StepAudit:
...
if self._judge:
judged = await self._judge(prompt, event)
+ score = float(cast(float, judged.get("score", 0.5)))
audit = StepAudit(
step_index=step_index,
event_id=event.event_id,
- score=float(cast(float, judged.get("score", 0.5))),
- verdict=str(judged.get("verdict", "uncertain")),
+ score=score,
+ verdict=self._advisory_verdict(score),
...
)
...
- if score >= ADVISORY_NOTE_THRESHOLD:
- verdict = ADVISORY_CLEAN
- elif score >= ADVISORY_WARN_THRESHOLD:
- verdict = ADVISORY_NOTE
- else:
- verdict = ADVISORY_WARN
+ verdict = self._advisory_verdict(score)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| # 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." | |
| # 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. | |
| verdict = self._advisory_verdict(score) | |
| rationale = ( | |
| "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." |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agentwatch/reasoning/auditor.py` around lines 330 - 342, Unify verdict
classification through a shared score-to-advisory-verdict helper, and use it for
both heuristic and judge-backed audits so StepAudit.verdict always emits
advisory values when is_advisory_only is true. Update the judge path around its
existing verdict handling to retain any judge-specific label separately if
needed, while preserving the score-based advisory result; add a regression test
covering a configured judge returning a non-advisory verdict.
Comment on lines
+221
to
+232
| 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" |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Assert the exact delegation result.
The loop accepts duplicate or stale structural signals. This event has no outcome-only signal, so audit.risk_signals must contain exactly the Capability Lattice names.
Proposed fix
- for name in expected:
- assert name in audit.risk_signals, f"{name!r} missing from auditor's risk_signals"
+ assert set(audit.risk_signals) == set(expected)
+ assert len(audit.risk_signals) == len(expected)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| 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_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() | |
| assert set(audit.risk_signals) == set(expected) | |
| assert len(audit.risk_signals) == len(expected) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_auditor_advisory.py` around lines 221 - 232, Update
test_auditor_risk_signals_match_capability_lattice_names to assert that
audit.risk_signals exactly equals
detect_capability_signals(event).to_risk_signal_names(), preserving the expected
ordering and rejecting duplicates or stale signals instead of checking
membership individually.
Comment on lines
+243
to
+246
| 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" | ||
| ) |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Assert that the event status is unchanged.
The current assertion permits a transition from RUNNING to any non-BLOCKED status. Capture the initial status and assert equality after audit_step().
Proposed fix
+ original_status = event.status
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"
+ assert event.status == original_status, (
+ "Phase 5 (`#665`): auditor is advisory-only and must not modify event status"
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| 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" | |
| ) | |
| original_status = event.status | |
| asyncio.run(auditor.audit_step(0, event)) | |
| assert event.status == original_status, ( | |
| "Phase 5 (`#665`): auditor is advisory-only and must not modify event status" | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_auditor_advisory.py` around lines 243 - 246, Update the test
around auditor.audit_step() to capture event.status before invocation, then
assert the status is equal to that captured value afterward. Replace the current
non-BLOCKED assertion while preserving the existing advisory-only test context.
SHAURYASANYAL3
approved these changes
Aug 11, 2026
SHAURYASANYAL3
left a comment
SHAURYASANYAL3
left a comment
Collaborator
There was a problem hiding this comment.
Deprecating the v1 auditor is exactly what we needed. Clean and precise.
Collaborator
|
I tried to merge this, but there are merge conflicts. Please resolve the conflicts and update the PR so I can merge it. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Promotes the v1
ReasoningAuditorto advisory-only as specified in Phase 5 of the v2 Cognitive Lattice (#665):_DESTRUCTIVE_PATTERNSsubstring matching out ofauditor.pyinto a newagentwatch.lattice.capabilitymodule (Capability Lattice).sound/acceptable/weak→advisory_clean/advisory_note/advisory_warnso consumers can distinguish an advisory warning from a block signal at a glance.StepAudit.is_advisory_only = Trueboolean (new field, defaults True) and surfaces it into_dict()so downstream consumers know the auditor never participates in blocking decisions.Why no
SafetyEnginechangeThe
SafetyEnginealready does not consume the auditor's verdict to make block decisions (agentwatch/core/safety.py:629-636). The auditor storesConfidenceDataon the event; blocking is decided entirely byRiskScorer+ policy DSL. Therefore no change toSafetyEnginewas needed — the legacy comment strings insafety.pyreferring to the auditor's confidence scoring are still accurate. The legacy regex-based pattern matching that did live in the auditor is now owned by the Capability Lattice.Files
agentwatch/lattice/capability.py(NEW)CapabilitySignal+CapabilitySignals(frozen dataclasses)detect_capability_signals(event) -> CapabilitySignals— structural signal detectorCapabilitySignals.to_risk_signal_names() -> list[str]— back-compat surface for v1 dashboardsagentwatch/lattice/__init__.py— exports new symbolsagentwatch/lattice/attention_scatter.py— also included (closes [v2] Cognitive Lattice: Build Attention Scatter Detector #662 dependency)agentwatch/reasoning/auditor.pyis_advisory_only: bool = Truefield onStepAuditdetect_risk_signalsnow delegates to Capability Lattice_heuristic_auditverdict uses new taxonomytests/test_auditor_advisory.py(NEW) — 18 tests pinning the new advisory surfacetests/test_attention_scatter.py(NEW) — 24 tests for the scatter detectorLegacy compatibility
audit.risk_signalsstill emits"destructive_command","privilege_escalation","broad_wildcard","external_fetch","tool_error","blocked_action"— same set as v1.StepAudit.verdictis a free-formstr, so renaming the values is safe for any consumer using string equality (no schema migration needed)."sound"/"acceptable"/"weak") are exported asLEGACY_VERDICT_*constants so existing import paths keep working.Verification
ruff check agentwatch/lattice/ agentwatch/reasoning/auditor.py tests/test_auditor_advisory.py tests/test_attention_scatter.py— ✓ All checks passedruff format --check …— ✓ already formattedpytest tests/test_auditor_transparency.py tests/test_auditor_advisory.py tests/test_attention_scatter.py tests/test_shadow_filesystem.py— 83 passed in 0.41stest_auditor_transparency.py(5 tests) all still pass.Closes #665
Summary by CodeRabbit
New Features
Behavior Changes
Tests