[AAASM-6352] 🐛 (core): Deny redact verdict under enforce instead of allowing it

agent_assembly/core/assembly.py

@@ -45,7 +45,8 @@
 EnforcementMode = Literal["enforce", "observe", "disabled"]
 """Posture the governance gateway should apply to this agent's actions.
 
-* ``"enforce"`` — default; deny blocks the action, redact strips secrets.
+* ``"enforce"`` — default; deny blocks the action; a redact verdict also blocks it, because
+  the SDK cannot apply redaction client-side (AAASM-6352, AAASM-1920).
 * ``"observe"`` — dry-run; the gateway records what *would* have happened
   but lets every action through. Surfaced by ``aa audit list --dry-run-only``.
 * ``"disabled"`` — policy evaluation skipped entirely. Hermetic test only.

agent_assembly/core/runtime_interceptor.py

@@ -9,8 +9,9 @@
 :class:`RuntimeQueryInterceptor` wraps the ``GatewayClient`` and adds a
 ``check_tool_start`` that asks a connected
 ``agent_assembly._core.RuntimeClient`` whether the tool call is allowed, via
-``query_policy``. The runtime is the authority: it redacts in place, so this
-layer only ever *blocks* on an explicit ``deny`` and otherwise proceeds.
+``query_policy``. The native query returns only ``(decision, reason)``; no
+redacted payload crosses the ABI, so this layer cannot apply a ``redact``
+verdict client-side. Under enforce a ``redact`` therefore denies (AAASM-6352).
 
 Failure posture is governed by ``enforcement_mode`` (AAASM-3106):
 
@@ -30,8 +31,10 @@
   returns the bare client unchanged, a raising or error ``query_policy``
   proceeds, exactly as before.
 
-The runtime remains the authority on *redaction* (it redacts in place); this
-layer only ever decides allow / deny / pending.
+This layer cannot redact: a ``redact`` verdict would otherwise run the tool with
+the original, secret-bearing arguments, so under enforce it denies until redacted
+payload support lands (AAASM-1920, AAASM-6352). Under observe / disabled it
+proceeds (dry-run fail-open).
 """
 
 from __future__ import annotations
@@ -66,8 +69,16 @@
 # else — the proto3 zero value ``unspecified`` ("no decision rendered"), an unknown
 # string, an empty value, or a missing ``decision`` key — is not an authoritative
 # allow and must fail closed under ``enforce`` (AAASM-4014, AAASM-4166), matching the
-# Node SDK; the runtime remains the authority on redaction, so ``redact`` proceeds.
-_ALLOW_DECISIONS = frozenset({"allow", "redact"})
+# Node SDK. ``redact`` is deliberately NOT here: the native query returns no
+# redacted payload, so the tool would run with the original secret-bearing args
+# (AAASM-6352); it is handled explicitly in ``check_tool_start``.
+_ALLOW_DECISIONS = frozenset({"allow"})
+
+# Deny reason for a ``redact`` verdict this SDK cannot honour client-side (AAASM-6352).
+_REDACT_UNSUPPORTED_REASON = (
+    "gateway required redaction the SDK cannot apply client-side; "
+    "redacted payload support tracked in AAASM-1920/AAASM-6352"
+)
 
 # Deny reason surfaced when the native ``agent_assembly._core`` extension is absent
 # under an enforce posture (AAASM-4760). Previously this case returned the bare
@@ -341,8 +352,10 @@ def check_tool_start(
         * ``"deny"`` → ``{"status": "deny", "reason": ...}``.
         * ``"pending"`` → ``{"status": "pending", "reason": ...}`` so the
           adapter's existing approval path runs.
-        * ``"allow"`` / ``"redact"`` → ``{"status": "allow"}``. The runtime
-          redacts authoritatively; this layer never redacts.
+        * ``"allow"`` → ``{"status": "allow"}``.
+        * ``"redact"`` → ``deny`` under ``enforce`` (no redacted payload crosses
+          the ABI, so the tool would run with the original secrets; AAASM-6352),
+          else ``allow`` (dry-run fail-open).
         * A raising ``query_policy``, the proto3 zero value ``"unspecified"``
           ("no decision rendered"), an error-sentinel ``decision``
           (``query_failed`` / ``channel_closed`` / ``shutdown``), or any
@@ -395,6 +408,8 @@ def check_tool_start(
         if decision in _ERROR_DECISIONS:
             # Native reported it could not reach an authoritative verdict.
             return self._on_query_failure(reason or f"runtime returned {decision}")
+        if decision == "redact":
+            return self._on_query_failure(_REDACT_UNSUPPORTED_REASON)
         if decision in _ALLOW_DECISIONS:
             return {"status": "allow"}
         # Unknown / empty / missing decision: not an authoritative allow, so fail

test/integration/test_native_core_runtime.py

@@ -49,21 +49,34 @@ def _run(self) -> None:
             with conn:
                 conn.settimeout(0.2)
 
-                # Heartbeat frame from client is a single tag byte.
-                tag = self._read_u8(conn)
-                if tag != 4:
-                    return
-                self._write_frame(conn, 3, b"")
+                # AAASM-3587: the real runtime issues a handshake challenge first
+                # and the shared client completes it (fail-closed) before any
+                # other traffic. The mock needs only a decodable challenge
+                # (HandshakeChallenge { nonce: bytes } = field 1, 32 bytes); it
+                # does not verify the proof — that is the runtime's job.
+                self._write_frame(conn, 5, b"\x0a\x20" + b"\x5a" * 32)
 
                 while not self._stop.is_set():
                     try:
                         tag = self._read_u8(conn)
-                    except (TimeoutError, OSError):
+                    except TimeoutError:
+                        continue
+                    except OSError:
+                        return
+
+                    if tag == 4:
+                        # Heartbeat frame from client is a single tag byte.
+                        try:
+                            self._write_frame(conn, 3, b"")
+                        except OSError:
+                            return
                         continue
 
-                    if tag in (1, 2, 3):
+                    if tag in (1, 2, 3, 5):
                         _ = self._read_length_delimited(conn)
 
+                    if tag == 5:
+                        continue
                     if tag == 1:
                         if self._policy_delay_ms > 0:
                             time.sleep(self._policy_delay_ms / 1000.0)
@@ -237,6 +250,54 @@ def test_query_policy_returns_deny_when_runtime_denies(native_core: Any) -> None
         server.close()
 
 
+def _redact_check(native_core: Any, payload: bytes, *, enforce: bool) -> dict[str, Any]:
+    """Drive the REAL native client + production interceptor against a mock runtime."""
+    from agent_assembly.core.runtime_interceptor import RuntimeQueryInterceptor
+
+    server = MockRuntimeServer(policy_payload=payload)
+    server.start()
+    client = native_core.RuntimeClient.connect(server.socket_path)
+    try:
+        interceptor = RuntimeQueryInterceptor(object(), client, "agent-001", enforce=enforce)
+        return interceptor.check_tool_start(serialized={"name": "web_search"}, input_str="query")
+    finally:
+        client.close()
+        server.close()
+
+
+@pytest.mark.integration
+def test_native_surfaces_a_runtime_redact_verdict(native_core: Any) -> None:
+    # CheckActionResponse { decision: REDACT } => field 1 (varint) = 4.
+    server = MockRuntimeServer(policy_payload=b"\x08\x04")
+    server.start()
+    client = native_core.RuntimeClient.connect(server.socket_path)
+    try:
+        result = client.query_policy("agent-001", "tool_call", "web_search", '{"q": "x"}')
+        assert result["decision"] == "redact"
+    finally:
+        client.close()
+        server.close()
+
+
+@pytest.mark.integration
+def test_real_native_redact_denies_under_enforce(native_core: Any) -> None:
+    """AAASM-6352: a real runtime REDACT must not run the tool with the original args."""
+    result = _redact_check(native_core, b"\x08\x04", enforce=True)
+    assert result["status"] == "deny"
+    assert "cannot apply client-side" in result["reason"]
+
+
+@pytest.mark.integration
+def test_real_native_allow_proceeds_under_enforce_control(native_core: Any) -> None:
+    """Paired control: same path and posture, only the runtime verdict differs."""
+    assert _redact_check(native_core, b"\x08\x01", enforce=True) == {"status": "allow"}
+
+
+@pytest.mark.integration
+def test_real_native_redact_stays_advisory_without_enforce(native_core: Any) -> None:
+    assert _redact_check(native_core, b"\x08\x04", enforce=False) == {"status": "allow"}
+
+
 @pytest.mark.integration
 def test_query_policy_fails_open_when_runtime_unreachable(native_core: Any) -> None:
     # No server bound at this path: the runtime never answers, so the shared

test/unit/adapters/test_redact_verdict_not_allow.py

@@ -0,0 +1,111 @@
+"""AAASM-6352: a ``redact`` verdict must not let the tool run under enforce.
+
+The native query returns only ``(decision, reason)``; no redacted payload crosses
+the ABI, so a tool run after ``redact`` would receive the ORIGINAL, secret-bearing
+arguments. These tests drive the real ``RuntimeQueryInterceptor`` through two
+adapter paths that consume ``check_tool_start`` and assert on what the tool
+actually receives, each paired with an ``allow`` control on the same flow.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from uuid import uuid4
+
+import pytest
+
+from agent_assembly.adapters._shared.tool_governance import run_governed_async_tool
+from agent_assembly.adapters.langchain import AssemblyCallbackHandler
+from agent_assembly.core.runtime_interceptor import RuntimeQueryInterceptor
+from agent_assembly.exceptions import PolicyViolationError, ToolExecutionBlockedError
+
+_SECRET = {"api_key": "sk-live-SECRET"}
+
+
+class _Runtime:
+    def __init__(self, decision: str) -> None:
+        self._decision = decision
+
+    def query_policy(self, *_args: Any, **_kwargs: Any) -> dict[str, str]:
+        return {"decision": self._decision, "reason": "r"}
+
+
+class _Gateway:
+    agent_id = "agent-001"
+
+
+def _interceptor(decision: str, *, enforce: bool) -> RuntimeQueryInterceptor:
+    return RuntimeQueryInterceptor(_Gateway(), _Runtime(decision), "agent-001", enforce=enforce)
+
+
+def _run_shared(decision: str, *, enforce: bool) -> list[dict[str, Any]]:
+    """Run the shared governed-tool flow; return the args each tool invocation received."""
+    received: list[dict[str, Any]] = []
+
+    def tool() -> str:
+        received.append(dict(_SECRET))
+        return "ok"
+
+    asyncio.run(
+        run_governed_async_tool(
+            _interceptor(decision, enforce=enforce),
+            enforce=enforce,
+            tool_name="t",
+            tool_args=dict(_SECRET),
+            agent_id="agent-001",
+            run_id="run-1",
+            invoke_original=tool,
+        )
+    )
+    return received
+
+
+def test_shared_path_redact_under_enforce_does_not_invoke_tool() -> None:
+    received: list[dict[str, Any]] = []
+
+    def tool() -> str:
+        received.append(dict(_SECRET))
+        return "ok"
+
+    with pytest.raises(PolicyViolationError):
+        asyncio.run(
+            run_governed_async_tool(
+                _interceptor("redact", enforce=True),
+                enforce=True,
+                tool_name="t",
+                tool_args=dict(_SECRET),
+                agent_id="agent-001",
+                run_id="run-1",
+                invoke_original=tool,
+            )
+        )
+
+    assert received == []
+
+
+def test_shared_path_allow_control_invokes_tool_with_args() -> None:
+    assert _run_shared("allow", enforce=True) == [_SECRET]
+
+
+def test_shared_path_redact_under_observe_still_runs_tool() -> None:
+    assert _run_shared("redact", enforce=False) == [_SECRET]
+
+
+def _langchain_start(decision: str, *, enforce: bool) -> None:
+    AssemblyCallbackHandler(_interceptor(decision, enforce=enforce)).on_tool_start(
+        serialized={"name": "web_search"}, input_str="q", run_id=uuid4()
+    )
+
+
+def test_langchain_redact_under_enforce_blocks_tool_start() -> None:
+    with pytest.raises(ToolExecutionBlockedError):
+        _langchain_start("redact", enforce=True)
+
+
+def test_langchain_allow_control_does_not_block() -> None:
+    _langchain_start("allow", enforce=True)
+
+
+def test_langchain_redact_under_observe_does_not_block() -> None:
+    _langchain_start("redact", enforce=False)

test/unit/core/test_runtime_interceptor.py

@@ -74,7 +74,7 @@ def test_deny_decision_maps_to_block_status() -> None:
     assert runtime_client.calls == [("agent-001", "tool_call", "web_search", '{"q": "x"}')]
 
 
-@pytest.mark.parametrize("decision", ["allow", "redact", "unspecified", "anything-else"])
+@pytest.mark.parametrize("decision", ["allow", "unspecified", "anything-else"])
 def test_non_deny_decisions_proceed(decision: str) -> None:
     interceptor = RuntimeQueryInterceptor(_FakeGatewayClient(), _FakeRuntimeClient(decision), "agent-001")
 
@@ -752,12 +752,35 @@ def query_policy(self, *_args: Any, **_kwargs: Any) -> dict[str, str]:
     assert result["status"] == "deny"
 
 
-@pytest.mark.parametrize("decision", ["allow", "redact"])
-def test_known_good_decisions_allow_under_enforce(decision: str) -> None:
-    """Authoritative allow verdicts still proceed under enforce; the runtime
-    remains the authority on redaction, so ``redact`` proceeds here. ``unspecified``
-    is no longer among them — it fails closed (AAASM-4166)."""
-    interceptor = RuntimeQueryInterceptor(_FakeGatewayClient(), _FakeRuntimeClient(decision), "agent-001", enforce=True)
+def test_allow_decision_proceeds_under_enforce() -> None:
+    """An authoritative ``allow`` still proceeds under enforce. ``unspecified``
+    is not one — it fails closed (AAASM-4166) — and neither is ``redact``
+    (AAASM-6352): the SDK cannot apply it, so it is covered by the redact tests."""
+    interceptor = RuntimeQueryInterceptor(_FakeGatewayClient(), _FakeRuntimeClient("allow"), "agent-001", enforce=True)
+
+    result = interceptor.check_tool_start(serialized={"name": "t"}, input_str="i")
+
+    assert result == {"status": "allow"}
+
+
+def test_redact_decision_denies_under_enforce() -> None:
+    """AAASM-6352: no redacted payload crosses the ABI, so allowing ``redact``
+    would run the tool with the original secret-bearing args. Deny, with a reason
+    naming the limitation."""
+    interceptor = RuntimeQueryInterceptor(_FakeGatewayClient(), _FakeRuntimeClient("redact"), "agent-001", enforce=True)
+
+    result = interceptor.check_tool_start(serialized={"name": "t"}, input_str="i")
+
+    assert result["status"] == "deny"
+    assert "cannot apply client-side" in result["reason"]
+    assert "AAASM-1920" in result["reason"]
+
+
+def test_redact_decision_allows_under_observe() -> None:
+    """Observe / disabled keep the dry-run fail-open posture for ``redact``."""
+    interceptor = RuntimeQueryInterceptor(
+        _FakeGatewayClient(), _FakeRuntimeClient("redact"), "agent-001", enforce=False
+    )
 
     result = interceptor.check_tool_start(serialized={"name": "t"}, input_str="i")