diff --git a/docs/acs.md b/docs/acs.md index 546d7abb..8ff96bf9 100644 --- a/docs/acs.md +++ b/docs/acs.md @@ -174,149 +174,77 @@ See [Conformance Profiles](./spec/conformance.md) for what each profile requires **Standards**: [ACS Instrument](./spec/instrument/README.md). - === "LangChain (Python)" - - ```python - from acs import GuardianAgent, PolicyResponse - - # Create a Guardian Agent to enforce policies - guardian = GuardianAgent( - endpoint="https://guardian.example.com", - policies=["data-protection", "cost-control"] - ) - - # Hook into agent runtime - @acs_agent.hook("toolCallRequest") - async def on_tool_call(context): - # Guardian evaluates the tool call - response = await guardian.evaluate({ - "hook": "toolCallRequest", - "tool": context.tool_name, - "parameters": context.parameters, - "session": context.session_id - }) - - if response.action == "DENY": - raise PermissionError(f"Tool call denied: {response.reason}") - elif response.action == "MODIFY": - # Apply Guardian's modifications - context.parameters = response.modifications - - return response - - # Example: Guardian denies sensitive data access - result = await acs_agent.run("Search for employee SSN records") - # > PermissionError: Tool call denied: Accessing PII data violates policy - ``` - - === "Vercel AI (TypeScript)" - - ```typescript - import { GuardianAgent, HookContext } from '@acs/guardian' - - // Configure Guardian Agent - const guardian = new GuardianAgent({ - endpoint: 'https://guardian.example.com', - policies: ['data-protection', 'cost-control'] - }) - - // Register hooks for runtime control - acsAgent.registerHook('toolCallRequest', async (context: HookContext) => { - const response = await guardian.evaluate({ - hook: 'toolCallRequest', - tool: context.toolName, - parameters: context.parameters, - session: context.sessionId - }) - - switch (response.action) { - case 'DENY': - throw new Error(`Denied: ${response.reason}`) - case 'MODIFY': - context.parameters = response.modifications - break - } - - return response - }) - - // Example: Cost control policy in action - await acsAgent.run('Generate 1000 images using DALL-E') - // > Error: Denied: Request exceeds cost threshold ($50 limit) - ``` + === "ACS wire (JSON-RPC)" - === "MCP" + Before a tool runs, the agent sends the call to the Guardian as [`steps/toolCallRequest`](./spec/instrument/hooks.md#toolcallrequest) and waits for the decision. Here the agent wants to run `SELECT * FROM customers`: ```json { "jsonrpc": "2.0", - "method": "acs/registerHook", + "method": "steps/toolCallRequest", + "id": "rpc-42", "params": { - "hook": "toolCallRequest", - "guardianEndpoint": "https://guardian.example.com/evaluate", - "policies": ["data-protection", "cost-control"] - } - } - - // Example hook evaluation request from Guardian - { - "jsonrpc": "2.0", - "method": "guardian/evaluate", - "params": { - "hook": "toolCallRequest", - "context": { - "tool": "database_query", - "parameters": { - "query": "SELECT * FROM customers" - }, - "session": "sess_123" + "acs_version": "0.1.0", + "request_id": "c329d5a4-97ae-44c8-986d-77d1da194ab7", + "timestamp": "2026-10-13T09:15:00Z", + "metadata": { + "agent_id": "research-assistant", + "session_id": "17a301f1-6d17-4941-8bea-92320f0773aa" + }, + "payload": { + "tool": { "name": "database_query" }, + "capability": "datastore.read", + "arguments": { + "query": { "value": "SELECT * FROM customers" } + } + }, + "signature": { + "algorithm": "HMAC-SHA256", + "value": "", + "key_id": "acs-hmac-key-1" } } } - - // Guardian response + ``` + + The Guardian answers `modify`. Its `parameter_overrides` replaces the `query` argument, so the agent runs the narrowed query instead of the original: + + ```json { "jsonrpc": "2.0", + "id": "rpc-42", "result": { - "action": "MODIFY", + "type": "final", + "acs_version": "0.1.0", + "request_id": "c329d5a4-97ae-44c8-986d-77d1da194ab7", + "decision": "modify", + "reasoning": "Narrowed the query to two columns and 100 rows. A full customer export violates the data-protection policy.", + "reason_codes": ["pii_detected"], + "policy_references": [ + { "policy_id": "data-protection", "rule_id": "limit-customer-export" } + ], "modifications": { - "query": "SELECT id, name FROM customers LIMIT 100" + "parameter_overrides": { + "query": "SELECT id, name FROM customers LIMIT 100" + } }, - "reason": "Limited query to prevent full data exposure" + "chain_hash": "25d878e6726ecf3bd9ee823d7642bdd78f148c0db5c06a585d16bc23494e2175", + "metadata": { "evaluator": "deterministic" }, + "signature": { + "algorithm": "HMAC-SHA256", + "value": "", + "key_id": "acs-hmac-key-1" + } } } ``` - === "A2A" + The exchange follows four rules: - ```yaml - # Available ACS hooks for instrumentation - - agentTrigger: - description: Fires when agent starts processing - can_modify: true - can_deny: true - - message: - description: Fires for each message in conversation - can_modify: true - can_deny: true - - toolCallRequest: - description: Fires before tool execution - can_modify: true - can_deny: true - - knowledgeRetrieval: - description: Fires when accessing knowledge bases - can_modify: false - can_deny: true - - memoryStore: - description: Fires when storing to memory - can_modify: true - can_deny: true - ``` + - **Signatures.** ACS-Core requires signatures on this request and response. Remove `params.signature` from the request or `result.signature` from the response, canonicalize the remaining envelope with JCS, and encode it as UTF-8 before signing ([§10](./spec/instrument/specification.md#10-cryptographic-signatures)). The `value` fields above are placeholders. The identifiers and `chain_hash` are sample values. + - **Chain head.** The Guardian publishes `chain_hash` on every content-bearing step, so an observer can detect a rewritten audit chain ([§8.6](./spec/instrument/specification.md#86-chain-head-publication-normative)). + - **MODIFY support.** `modify` is SHOULD-support. A client that doesn't implement it receives `deny` with `reason_codes: ["modify_unsupported"]` instead ([§6.5](./spec/instrument/specification.md#65-modify-incapable-clients-normative)). + - **Protocols.** An MCP `tools/call` can travel through this same hook ([Instrument MCP](./spec/instrument/extend_mcp.md)). A2A wrapping is reserved for v0.2. ### Traceable diff --git a/tests/test_acs_examples.py b/tests/test_acs_examples.py new file mode 100644 index 00000000..8547aa6d --- /dev/null +++ b/tests/test_acs_examples.py @@ -0,0 +1,216 @@ +"""Keep the wire examples on acs.md valid against the v0.1.0 schemas. + +The overview page shipped requests for methods ACS never defined +(`acs/registerHook`, `guardian/evaluate`) and a MODIFY response the response +envelope rejects. The build and link guards accepted them, because nothing read +the JSON. Read it here. + +Envelope validation alone is not enough. The envelope's `method` pattern checks +only the namespace prefix, so `steps/anything` passes it, and the envelope +accepts any object as `payload`. A request therefore has to name a method whose +payload schema exists, and its payload has to satisfy that schema. A response +has to fit the response envelope and answer a request shown beside it. Every +request is checked before indexing by ID, and duplicate IDs are rejected so +one request cannot hide another. + +The schemas leave signatures and chain heads field-optional. Check their +required presence for these Core examples too. Signature and chain values +are illustrative; this guard does not verify cryptographic authenticity. + +Sections join SECTIONS as their examples are rebuilt (#267). Every JSON block in +a listed section must be a JSON-RPC message; a section that needs other JSON +needs its own rule. +""" + +import json +import re +from copy import deepcopy +from pathlib import Path + +import pytest +from jsonschema import Draft202012Validator, FormatChecker, ValidationError +from referencing import Registry, Resource +from referencing.jsonschema import DRAFT202012 + +ROOT = Path(__file__).resolve().parent.parent +SPEC = ROOT / "specification" / "v0.1.0" +PAGE = ROOT / "docs" / "acs.md" +SECTIONS = ["Instrumentable"] + + +def load_schemas(): + schemas = {} + for path in sorted(SPEC.rglob("*.json")): + doc = json.loads(path.read_text(encoding="utf-8")) + if isinstance(doc, dict) and "$id" in doc: + schemas[path.relative_to(SPEC).as_posix()] = doc + return schemas + + +SCHEMAS = load_schemas() +REGISTRY = Registry().with_resources( + (doc["$id"], Resource.from_contents(doc, default_specification=DRAFT202012)) + for doc in SCHEMAS.values() +) + +# The base payload schema for each defined method, read from its title +# ("steps/toolCallRequest payload"). Profile variants carry a suffix and are +# left out, so a Core example is checked against the Core schema. +PAYLOADS = { + match[1]: doc + for doc in SCHEMAS.values() + if (match := re.fullmatch(r"(\S+) payload", doc.get("title", ""))) +} + + +def validate(schema, instance): + Draft202012Validator(schema, registry=REGISTRY, format_checker=FormatChecker()).validate(instance) + + +def section(name): + text = PAGE.read_text(encoding="utf-8") + start = text.index(f"\n### {name}\n") + following = re.search(r"\n#{1,3} ", text[start + 1:]) + return text[start:start + 1 + following.start()] if following else text[start:] + + +def messages(name): + blocks = re.findall(r"```json\n(.*?)```", section(name), re.DOTALL) + shown = [json.loads(block) for block in blocks] + for message in shown: + assert message.get("jsonrpc") == "2.0", f"{name} shows JSON that is not a JSON-RPC message" + return shown + + +def check_request(request): + method = request["method"] + assert method in PAYLOADS, f"{method} is not a defined v0.1 method" + validate(SCHEMAS["request-envelope.json"], request) + validate(PAYLOADS[method], request["params"]["payload"]) + # system/ping is explicitly exempt from Core signature requirements. + if method != "system/ping": + assert "signature" in request["params"], "ACS-Core requests require params.signature" + + +def check_response(response, requests): + validate(SCHEMAS["response-envelope.json"], response) + request = requests.get(response["id"]) + assert request, f"Response {response['id']} answers no request shown beside it" + result = response["result"] + assert result["request_id"] == request["params"]["request_id"] + if request["method"] != "system/ping": + assert "signature" in result, "ACS-Core responses require result.signature" + # This rebuilt example is a content-bearing hook (Specification §8.6). + # Extend this check when other content-bearing examples join SECTIONS. + if request["method"] == "steps/toolCallRequest": + assert "chain_hash" in result, "toolCallRequest responses require result.chain_hash" + overrides = result.get("modifications", {}).get("parameter_overrides", {}) + arguments = request["params"]["payload"].get("arguments", {}) + assert set(overrides) <= set(arguments), "parameter_overrides names an argument the request does not carry" + + +def split(name): + shown = messages(name) + requests = {} + responses = [] + for message in shown: + if "method" in message: + check_request(message) + assert message["id"] not in requests, f"{name} repeats request ID {message['id']}" + requests[message["id"]] = message + else: + responses.append(message) + return requests, responses + + +@pytest.mark.parametrize("name", SECTIONS) +def test_requests_use_defined_methods_and_valid_payloads(name): + requests, _ = split(name) + assert requests, f"{name} shows no request" + + +@pytest.mark.parametrize("name", SECTIONS) +def test_responses_answer_a_request_shown_beside_them(name): + requests, responses = split(name) + assert responses, f"{name} shows no response" + for response in responses: + check_response(response, requests) + + +def instrumentable_exchange(): + requests, responses = split("Instrumentable") + return deepcopy(requests), deepcopy(responses[0]) + + +@pytest.mark.parametrize("method", ["acs/registerHook", "guardian/evaluate", "steps/registerHook"]) +def test_undefined_methods_are_rejected(method): + requests, _ = instrumentable_exchange() + request = next(iter(requests.values())) + request["method"] = method + with pytest.raises((AssertionError, ValidationError)): + check_request(request) + + +def test_a_bare_argument_value_is_rejected(): + requests, _ = instrumentable_exchange() + request = next(iter(requests.values())) + request["params"]["payload"]["arguments"]["query"] = "SELECT * FROM customers" + with pytest.raises(ValidationError): + check_request(request) + + +def test_an_unsigned_core_request_is_rejected(): + requests, _ = instrumentable_exchange() + request = next(iter(requests.values())) + request["params"].pop("signature") + # This is schema-valid; Core imposes the stronger presence requirement. + validate(SCHEMAS["request-envelope.json"], request) + with pytest.raises(AssertionError, match=r"require params\.signature"): + check_request(request) + + +@pytest.mark.parametrize("field", ["signature", "chain_hash"]) +def test_required_core_response_fields_cannot_be_omitted(field): + requests, response = instrumentable_exchange() + response["result"].pop(field) + validate(SCHEMAS["response-envelope.json"], response) + with pytest.raises(AssertionError, match=rf"require result\.{field}"): + check_response(response, requests) + + +@pytest.mark.parametrize("invalid_first", [True, False], ids=["invalid-first", "invalid-last"]) +def test_a_request_with_a_reused_id_cannot_hide_an_invalid_method(monkeypatch, invalid_first): + requests, response = instrumentable_exchange() + valid = next(iter(requests.values())) + invalid = deepcopy(valid) + invalid["method"] = "steps/notDefined" + shown = [invalid, valid, response] if invalid_first else [valid, invalid, response] + monkeypatch.setattr(f"{__name__}.messages", lambda _: shown) + with pytest.raises(AssertionError, match="steps/notDefined is not a defined v0.1 method"): + split("Instrumentable") + + +def test_duplicate_valid_request_ids_are_rejected(monkeypatch): + requests, response = instrumentable_exchange() + request = next(iter(requests.values())) + shown = [request, deepcopy(request), response] + monkeypatch.setattr(f"{__name__}.messages", lambda _: shown) + with pytest.raises(AssertionError, match="repeats request ID"): + split("Instrumentable") + + +@pytest.mark.parametrize("mutation", [ + pytest.param(lambda result: result.pop("modifications"), id="modify-without-modifications"), + pytest.param(lambda result: result.pop("reasoning"), id="modify-without-reasoning"), + pytest.param(lambda result: result.update(decision="MODIFY"), id="uppercase-decision"), + pytest.param(lambda result: result.update(request_id="not-a-uuid"), id="non-uuid-request-id"), + pytest.param( + lambda result: result["modifications"].update(parameter_overrides={"table": "customers"}), + id="override-for-an-argument-not-sent", + ), +]) +def test_malformed_modify_responses_are_rejected(mutation): + requests, response = instrumentable_exchange() + mutation(response["result"]) + with pytest.raises((AssertionError, ValidationError)): + check_response(response, requests)