Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion .github/workflows/quickstart-tabs-check.yml
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@ on:
# these entries matter for the docs-only PR, which ci.yaml deliberately skips.
- "test/unit/test_quickstart_claim_bindings.py"
- "test/unit/test_quickstart_negative_control.py"
# AAASM-5661: the second control module the gate reads β€” the one that runs the
# documented configuration with no fake native core.
- "test/unit/test_quickstart_documented_configuration.py"
- "agent_assembly/exceptions/**"
push:
branches:
Expand Down Expand Up @@ -79,5 +82,13 @@ jobs:
- name: Install the SDK and its dev dependencies
run: uv sync

# AAASM-5661: the controls run here too, not only the bindings. A binding
# names a control by node id, and the gate checks the id still exists β€” it
# cannot tell whether that control still passes. On a docs-only PR ci.yaml
# does not run, so without this the named evidence went unexecuted.
- name: Every documented enforcement claim names the control that proves it
run: uv run pytest test/unit/test_quickstart_claim_bindings.py -q --no-cov
run: >-
uv run pytest
test/unit/test_quickstart_claim_bindings.py
test/unit/test_quickstart_documented_configuration.py
-q --no-cov
45 changes: 45 additions & 0 deletions agent_assembly/core/runtime_interceptor.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import json
import os
import stat
import sys
import warnings
from importlib import metadata
from typing import Any
Expand Down Expand Up @@ -118,6 +119,40 @@ def _warn_sdk_enforcement_unavailable() -> None:
)


def _warn_sdk_enforcement_not_applied(reason: str) -> None:
"""Warn (loudly) that this build applies no SDK-layer enforcement (AAASM-5661).

Emitted on the fail-open branch of :func:`_governance_unavailable`: an explicit
``observe`` / ``disabled`` posture with no authoritative runtime to consult. The
bare ``GatewayClient`` returned there exposes no ``check_tool_start``, so the
adapters' missing-interceptor fallback allows and the governed call runs β€” the
genuine no-op path, and the quiet one. ``init_assembly`` already warns about
*registration* on the same configuration, and that warning has been mistaken for
the whole story: an agent can be registered and still run with no in-process
allow/deny.

Written straight to ``sys.stderr`` rather than through ``warnings`` for the same
reason as :func:`~agent_assembly.core.assembly._warn_agent_unregistered`: a
``logging`` or ``warnings`` filter must not be able to silence a statement about
what the SDK is not doing. Once per interceptor build, so it cannot become
per-call noise.

:param reason: The clause naming why no authority was reachable, reused from the
deny reason the enforce branch would have carried (contains no credentials).
"""
sys.stderr.write(
"[agent-assembly] WARNING: SDK-layer enforcement is NOT applied on this path "
f"({reason}). Under enforcement_mode='observe' / 'disabled' the SDK stays "
"advisory, so a governed tool call reaches its body with no in-process "
"allow/deny decision behind it, and a policy DENY does not block it here. This "
"is a separate gap from the registration warning: an agent can be registered "
"and still run with no SDK-layer enforcement. Use the default enforce posture "
"with the native agent_assembly._core extension installed to obtain the "
"in-process decision; the proxy / eBPF layers remain authoritative either way "
"(AAASM-5661).\n"
)


def _resolve_runtime_socket_path(agent_id: str) -> str:
"""Resolve the runtime UDS path: ``AA_RUNTIME_SOCKET`` > default convention.

Expand Down Expand Up @@ -562,8 +597,18 @@ def _governance_unavailable(client: Any, enforce: bool, reason: str, *, warn: bo
``warn`` gates the one-time loud warning to the native-missing case (a
pure-Python install), matching the historical AAASM-4130 behavior; the
unreachable-socket case denies without an extra warning.

The fail-open branch warns unconditionally (AAASM-5661). It is the one
remaining path on which a governed tool call reaches its body with no
in-process decision behind it, and until now it was the quietest: the bare
``GatewayClient`` exposes no ``check_tool_start``, so the adapters fall back
to an allow and the session looks governed. The only loud signal a caller got
was :func:`~agent_assembly.core.assembly._warn_agent_unregistered`, which is
about *registration* β€” a reader who saw it and shrugged had no way to learn
that enforcement was gone too.
"""
if not enforce:
_warn_sdk_enforcement_not_applied(reason)
return client
if warn:
_warn_sdk_enforcement_unavailable()
Expand Down
53 changes: 40 additions & 13 deletions docs/quick-start.md
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
# Quick Start

Govern your first agent in about five minutes. By the end you'll have an agent β€” in whichever
framework you already use β€” whose tool calls pass through the Agent Assembly policy gate, and it
framework you already use β€” whose tool calls pass through an Agent Assembly adapter, and it
runs **offline** against a local policy, so you need no API keys and no network access to the
outside world.

Expand Down Expand Up @@ -666,27 +666,54 @@ with no API keys and no outbound network.

## What just happened

1. **`init_assembly()` wired in governance.** It registered the agent with the gateway and
auto-loaded the adapter for your framework β€” every tool call from this point on is routed
through the policy gate.
2. **`mode="sdk-only"` kept it offline.** The in-process adapter enforces on tool calls with no
network sidecar, so the example runs deterministically with no real LLM or gateway
round-trip.
3. **Tool calls were governed.** The adapter intercepts the framework's tool-invocation path and
asks the policy engine for an allow/deny verdict before the tool actually runs.
1. **`init_assembly()` installed the governance hook.** It attempted to register the agent with
the gateway and auto-loaded the adapter for your framework, which patches that framework's
tool-invocation path.
2. **`mode="sdk-only"` kept it offline.** No network sidecar starts in that mode, so the example
runs deterministically with no real LLM or gateway round-trip.
3. **Tool calls went through the adapter.** The adapter intercepts the framework's
tool-invocation path and, when a policy authority is reachable, asks it for an allow/deny
verdict before the tool actually runs.
4. **The `with` block tore everything down on exit** β€” adapter hooks were unwound and the
gateway connection closed, leaving the process exactly as it was before.

If a tool call raises a `ToolExecutionBlockedError`, that is not a bug β€” the policy denied the
call. That's the product working. See
If a tool call raises a `ToolExecutionBlockedError`, that is not a bug β€” something refused the
call before it ran. Read the exception's reason to see what refused it: a policy rule that denied
the call, or β€” as in this offline example β€” an SDK that had no authority to ask and refused
rather than run ungoverned. That's the product working. See
[Handling allow/deny decisions](guides/handling-decisions.md) for how to catch and respond to
those, and [Troubleshooting](troubleshooting.md) if `init_assembly()` itself raised.

## What this offline example evaluates

The example passes a `gateway_url`, and running it offline means nothing is listening there.
A pure-Python `{{ aa.commands.install_pip }}` carries no native `agent_assembly._core`
extension either.

`init_assembly()` reaches no policy authority in that configuration, so it evaluates no policy β€”
under
[ADR 0033 Β§6](https://github.com/ai-agent-assembly/agent-assembly/blob/master/docs/src/adr/0033-canonical-governance-and-enforcement-architecture.md)
the term for that state is **Degraded**, not *Evaluated*.
Under the default enforce posture the SDK takes its fail-closed branch instead: a governed tool
call is **denied before execution**, carrying a reason that names the absent extension rather
than a policy rule.
`init_assembly()` says as much at startup β€” it warns that the agent is unregistered, and that no
in-process policy decision can be made.

That is why several framework tabs above revert the hook `init_assembly()` installed and
re-apply one wired to the example's own `LocalPolicyEngine`.
The local engine, not the SDK, is what returns allow and deny in the offline demo.

Getting a decision from the SDK instead needs the native `agent_assembly._core` extension this
example lacks; install it with `{{ aa.commands.install_pip_runtime }}`.
[Point the SDK at a gateway](#2-point-the-sdk-at-a-gateway) above covers the other half of that
setup.

## `mode="sdk-only"` β€” why this example uses it

`mode="sdk-only"` is the in-process-only interception layer: the framework adapter enforces on
tool calls, with no network sidecar to start. It's the most portable mode and the best choice
for deterministic, offline examples and tests. The other modes (`auto`, `proxy`, `ebpf`) add
tool calls against a reachable policy authority, and starts no network sidecar. It's the most
portable mode and the best choice for deterministic, offline examples and tests. The other modes (`auto`, `proxy`, `ebpf`) add
network/kernel interception β€” see [Core Concepts β†’ Modes](concepts/index.md#runtime-modes).

## Next steps
Expand Down
78 changes: 75 additions & 3 deletions test/unit/core/test_runtime_interceptor.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,9 +266,18 @@
assert result.check_tool_start(serialized={"name": "t"}, input_str="i")["status"] == "deny"


def test_observe_mode_does_not_warn_when_native_core_missing(monkeypatch: pytest.MonkeyPatch) -> None:
"""The warning is scoped to the enforce posture: an explicit ``observe`` dry-run
with no native extension legitimately fails open and must stay silent."""
def test_observe_mode_reports_enforcement_is_not_applied_when_native_core_missing(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""Two separate contracts, and the test name used to assert only the first.

The enforce-specific ``UserWarning`` stays scoped to the enforce posture β€” an
explicit ``observe`` dry-run legitimately fails open, and telling that caller
their tools are being denied would be false. But silence about *enforcement*
was the AAASM-5661 gap: this branch hands back the bare client, whose missing
``check_tool_start`` makes the adapters allow, and the only loud signal the
caller got was about registration.
"""
monkeypatch.delitem(sys.modules, "agent_assembly._core", raising=False)

import builtins
Expand All @@ -289,6 +298,11 @@
result = build_governance_interceptor(client, "agent-001", "observe")

assert result is client
stderr = capsys.readouterr().err
assert "SDK-layer enforcement is NOT applied" in stderr
# The remedy has to be in the same breath as the gap; a caller who reads only
# the registration warning has no reason to look for a second one.
assert "agent_assembly._core" in stderr


def test_native_missing_deny_reason_is_explicit(monkeypatch: pytest.MonkeyPatch) -> None:
Expand All @@ -309,7 +323,7 @@

monkeypatch.setattr(builtins, "__import__", _no_core_import)

with pytest.warns(UserWarning, match="native runtime extension"):

Check warning on line 326 in test/unit/core/test_runtime_interceptor.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this warning test to have only one invocation possibly emitting a warning.

See more on https://sonarcloud.io/project/issues?id=ai-agent-assembly_python-sdk&issues=AZ_97KPerk564O0n07s6&open=AZ_97KPerk564O0n07s6&pullRequest=318
result = build_governance_interceptor(_FakeGatewayClient(), "agent-001", "enforce")

verdict = result.check_tool_start(serialized={"name": "t"}, input_str="i")
Expand Down Expand Up @@ -426,6 +440,64 @@
assert build_governance_interceptor(client, "agent-001", "observe") is client


def test_observe_unreachable_runtime_reports_enforcement_is_not_applied(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The second fail-open entry: native present, socket unreachable (AAASM-5661).

Distinct from the native-missing case above and previously the quieter of the
two β€” this branch never emitted anything at all, in either posture.
"""

class _UnreachableRuntimeClient:
@staticmethod
def connect(_socket_path: str) -> Any:
raise OSError("no such socket")

fake_core = types.ModuleType("agent_assembly._core")
fake_core.RuntimeClient = _UnreachableRuntimeClient # type: ignore[attr-defined]
monkeypatch.setitem(sys.modules, "agent_assembly._core", fake_core)

client = _FakeGatewayClient()
build_governance_interceptor(client, "agent-001", "observe")

stderr = capsys.readouterr().err
assert "SDK-layer enforcement is NOT applied" in stderr
assert "runtime unreachable" in stderr


def test_enforce_posture_does_not_report_enforcement_as_unapplied(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The notice is scoped to the fail-open branch β€” a positive control for it.

Without this, deleting the ``if not enforce`` guard and warning on every path
would leave the two controls above green while telling a fail-closed caller
that enforcement is not applied, which is the opposite of what happens.
"""
monkeypatch.delitem(sys.modules, "agent_assembly._core", raising=False)

import builtins
import warnings

real_import = builtins.__import__

def _no_core_import(name: str, *args: Any, **kwargs: Any) -> Any:
if name == "agent_assembly._core":
raise ImportError("native extension unavailable")
return real_import(name, *args, **kwargs)

monkeypatch.setattr(builtins, "__import__", _no_core_import)

client = _FakeGatewayClient()
with warnings.catch_warnings():
warnings.simplefilter("ignore")
result = build_governance_interceptor(client, "agent-001", "enforce")

assert isinstance(result, _FailClosedInterceptor)
assert "SDK-layer enforcement is NOT applied" not in capsys.readouterr().err


def test_enforce_wraps_with_fail_closed_query_path(monkeypatch: pytest.MonkeyPatch) -> None:
"""build_governance_interceptor under enforce wraps a reachable runtime so a
raising query denies (the wrapper carries enforce=True)."""
Expand Down
Loading