From 5bdbad6ec05816030cbd6ba08a09b1c9cf3d05f6 Mon Sep 17 00:00:00 2001 From: Cosmo Date: Wed, 14 Oct 2026 16:41:40 -0600 Subject: [PATCH] Add Anthropic SDK tool loop integration Records a tool loop written directly against the anthropic Python SDK as AgenTrust Telemetry events (policy decisions, actions, usage, data flows) and signs one log-import TRACE record per finished run. The in-process writer is standard library only and fail-open; the recorder runs out of process. Metadata only; unfinished runs are not signed. Includes an offline example, tests (writer, recorder, interop against the released anthropic SDK with a mocked transport), a path-scoped conformance workflow running trace-tests at Level 0, and the regenerated index and Marketplace catalog. Co-Authored-By: Claude Opus 5.5 --- .../anthropic-tool-loop-conformance.yml | 68 +++ README.md | 2 + integrations/anthropic-tool-loop/README.md | 214 +++++++++ .../anthropic_tool_loop_trace/__init__.py | 11 + .../anthropic_tool_loop_trace/catalog.py | 89 ++++ .../anthropic_tool_loop_trace/recorder.py | 451 ++++++++++++++++++ .../anthropic_tool_loop_trace/spool.py | 176 +++++++ .../examples/offline_tool_loop.py | 224 +++++++++ .../anthropic-tool-loop/integration.yaml | 22 + .../anthropic-tool-loop/pyproject.toml | 28 ++ .../tests/test_anthropic_interop.py | 91 ++++ .../tests/test_anthropic_recorder.py | 307 ++++++++++++ .../tests/test_anthropic_spool.py | 158 ++++++ marketplace/catalog.json | 26 +- 14 files changed, 1866 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/anthropic-tool-loop-conformance.yml create mode 100644 integrations/anthropic-tool-loop/README.md create mode 100644 integrations/anthropic-tool-loop/anthropic_tool_loop_trace/__init__.py create mode 100644 integrations/anthropic-tool-loop/anthropic_tool_loop_trace/catalog.py create mode 100644 integrations/anthropic-tool-loop/anthropic_tool_loop_trace/recorder.py create mode 100644 integrations/anthropic-tool-loop/anthropic_tool_loop_trace/spool.py create mode 100644 integrations/anthropic-tool-loop/examples/offline_tool_loop.py create mode 100644 integrations/anthropic-tool-loop/integration.yaml create mode 100644 integrations/anthropic-tool-loop/pyproject.toml create mode 100644 integrations/anthropic-tool-loop/tests/test_anthropic_interop.py create mode 100644 integrations/anthropic-tool-loop/tests/test_anthropic_recorder.py create mode 100644 integrations/anthropic-tool-loop/tests/test_anthropic_spool.py diff --git a/.github/workflows/anthropic-tool-loop-conformance.yml b/.github/workflows/anthropic-tool-loop-conformance.yml new file mode 100644 index 0000000..ee5f634 --- /dev/null +++ b/.github/workflows/anthropic-tool-loop-conformance.yml @@ -0,0 +1,68 @@ +# agentrust-io conformance workflow for the Anthropic SDK tool-loop integration. +# Lives in the repository root because GitHub Actions only discovers workflows +# in the root .github/workflows directory; scoped to this integration via paths. +name: anthropic-tool-loop conformance +on: + push: + paths: + - "integrations/anthropic-tool-loop/**" + - ".github/workflows/anthropic-tool-loop-conformance.yml" + pull_request: + paths: + - "integrations/anthropic-tool-loop/**" + - ".github/workflows/anthropic-tool-loop-conformance.yml" + schedule: + - cron: "0 6 * * 1" # weekly: catch drift against the latest released packages + workflow_dispatch: + +permissions: + contents: read + +jobs: + writer-py39: + # The in-process writer claims standard-library-only and Python 3.9 support. + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.9" + - name: Writer tests with nothing but pytest installed + run: | + pip install "pytest>=8" + pytest integrations/anthropic-tool-loop/tests/test_anthropic_spool.py -q + + conformance: + strategy: + fail-fast: false + matrix: + python: ["3.11", "3.12", "3.13"] + os: [ubuntu-latest] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python }} + - name: Install released agentrust-io packages and this integration + run: | + pip install agentrust-telemetry==0.1.0a6 agentrust-trace==0.11.0 agentrust-trace-tests==0.6.2 + pip install -e "integrations/anthropic-tool-loop[recorder,test]" + - name: Integration tests + run: pytest integrations/anthropic-tool-loop/tests -q + - name: Interop tests against the released anthropic SDK + run: | + pip install anthropic==1.14.0 + pytest integrations/anthropic-tool-loop/tests/test_anthropic_interop.py -q + - name: Offline example, emit a signed TRACE record + run: python integrations/anthropic-tool-loop/examples/offline_tool_loop.py --out trust-record.json + - name: TRACE conformance + run: trace-tests verify --record trust-record.json --level 0 + - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: conformance-${{ matrix.os }}-py${{ matrix.python }} + path: trust-record.json diff --git a/README.md b/README.md index a48fc12..86592d4 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,7 @@ TRACE only works as a standard if it is genuinely neutral. Integrations are list | [AAIF Authority Comparison Runner](integrations/alakris-authority-comparison/) | Alakris Research | trace | community | | [Agent Replay](integrations/altrudev-agent-replay/) | Altru.dev | trace | verified | | [Frequency Agent Execution Assurance](integrations/altrudev-frequency-agent-execution-assurance/) | Altru.dev | trace | community | +| [Anthropic SDK tool loop](integrations/anthropic-tool-loop/) | Disruptive Beverages | trace | community | | [Bernstein MCP verifier](integrations/bernstein-mcp/) | Bernstein | trace | verified | | [CHAP](integrations/chap/) | agentrust-io | trace | verified | | [AI Agent Incident Register](integrations/companyscope-incident-register/) | CompanyScope | agent-manifest | verified | @@ -105,6 +106,7 @@ adapter can and cannot see. | Framework | Adapter | Evidence source | Released framework exercised in CI | Evidence boundary | |---|---|---|---|---| +| Anthropic SDK (Messages API tool loop) | [Anthropic SDK tool loop](integrations/anthropic-tool-loop/) | Explicit calls in the application's own loop, read from a spool file (`log-import`) | Yes - anthropic 1.14.0 client with a mocked HTTP transport | Tool, model round, gate decision and input digest; caller-classified data flows; no prompts, arguments or results; no policy evaluation of its own | | Google ADK | [Google ADK](integrations/google-adk/) | First-party `BasePlugin` lifecycle | Yes - Google ADK 2.7.1 `InMemoryRunner` | Callback-visible invocation, model, and available tool identity; no payloads, retries, agent graph, function-body execution, or policy enforcement | | LangChain | [LangChain](integrations/langchain/) | First-party `BaseCallbackHandler` callbacks | Yes - LangChain Core 1.6.0 callback contract | Tool identity and outcome plus model identity; no chain topology or runnable state | | LangGraph | [LangChain](integrations/langchain/) | First-party LangChain callbacks propagated by the graph | Yes - LangGraph 1.2.11 `StateGraph` with a nested tool call | Propagated tool callbacks; no nodes, edges, state transitions, checkpoints, or rollback decisions | diff --git a/integrations/anthropic-tool-loop/README.md b/integrations/anthropic-tool-loop/README.md new file mode 100644 index 0000000..786632f --- /dev/null +++ b/integrations/anthropic-tool-loop/README.md @@ -0,0 +1,214 @@ +# Anthropic SDK tool loop → AgenTrust Telemetry and TRACE + +Records a tool-use loop written directly against the +[`anthropic`](https://pypi.org/project/anthropic/) Python SDK (the application +calls `client.messages.create` in a loop and dispatches each `tool_use` block +itself) as [AgenTrust Telemetry](https://github.com/agentrust-io/agentrust-telemetry) +governance events, and signs one TRACE v0.2 record per finished run. There is no +framework to hook into here, so the loop calls the recorder explicitly. + +The code this package was generalised from runs in production in front of two +Slack operations agents, an ERP/purchasing agent and a sales-operations agent. + +## Two parts + +| Part | Where it runs | Needs | +|---|---|---| +| `anthropic_tool_loop_trace.Run` | Inside the agent process | Standard library only, Python 3.9+ | +| `anthropic_tool_loop_trace.recorder.Recorder` | Out of process (sidecar, timer, cron) | Python 3.11+, `agentrust-telemetry`, `agentrust-trace`; `opentelemetry-sdk` for spans | + +`Run` appends metadata-only JSON lines to a spool file: `run.start`, one +`model.call` per `messages.create` round, one `tool.call` per `tool_use` block, +`run.end`. Every method is fail-open: a spool problem logs one warning and never +raises into the reply path. The agent process never imports the signing stack and +never holds the signing key. + +`Recorder.poll()` reads new rows, and for each finished run builds the events, +optional OpenTelemetry spans with the observed start and end times, an +`EvidenceAccumulator` chain, and a signed record from `finalize_trace` +(`origin.kind: log-import`). It never consumes a half-written line, and keeps its +read offset and unfinished runs in `state.json`, so a restart mid-run loses +nothing. + +## The hook lines + +```python +import time +from anthropic_tool_loop_trace import Run # 1 + +run = Run("spiffe://example.org/agent/purchasing", "purchasing", model, # 2 + toolset=[t["name"] for t in tools], action_field="action") +for round_no in range(max_rounds): + t0 = time.time_ns() + response = client.messages.create(model=model, max_tokens=1024, tools=tools, messages=messages) + run.model_call(round_no, t0, response) # 3 + if response.stop_reason != "tool_use": + run.end(response.stop_reason) # 4 + break + results = [] + for block in response.content: + if block.type != "tool_use": + continue + t0 = time.time_ns() + reason = my_gate(block.name, block.input) # your own check; None means allow + result = None if reason else my_tools[block.name](block.input) + run.tool_call(round_no, block, t0, decision="deny" if reason else "allow", # 5 + policy="my_gate", reason=reason, result=result) + results.append(...) + messages += [{"role": "assistant", "content": response.content}, + {"role": "user", "content": results}] +else: + run.end("max_rounds") # 6 +``` + +`tool_call` also takes `error_type=` when the tool raised and `gate_ns=` for how +long the check took. [`examples/offline_tool_loop.py`](examples/offline_tool_loop.py) +has the complete loop. + +## Recorder setup + +```python +from agentrust_trace import load_key +from anthropic_tool_loop_trace import ToolCatalog, ToolInfo +from anthropic_tool_loop_trace.recorder import Recorder, source_digest + +catalog = ToolCatalog( + { + "inventory_lookup": ToolInfo("inventory-db", "database", "internal"), + "purchase_orders": ToolInfo("erp", "erp", "confidential", + write=lambda verb: verb not in {"get", "list"}), + "crm_*": ToolInfo("crm", "crm", "confidential"), # prefix match + }, + taxonomy="example.v1", + order=("public", "internal", "confidential"), # lowest first +) +recorder = Recorder( + spool="~/.local/state/agentrust/anthropic-spool.jsonl", + data_dir="/var/lib/agent-records", + signing_key=load_key(open("recorder-key.pem").read()), + catalog=catalog, + enforcement_mode="enforce", # required, no default + policy_bundle_digest=source_digest("app/gates"), + build_digest=source_digest("app"), + prompt_classification="internal", # required: the recorder cannot see prompts + tracer_provider=my_tracer_provider, # optional +) +recorder.poll() # call on a timer; returns the run ids it recorded +``` + +Records land in `data_dir/records//.json` as +`{"run": summary, "events": [...], "trace": signed_record_or_null}`, the hash-chained +evidence in `data_dir/evidence/.jsonl`, and one summary line per run in +`data_dir/index.jsonl`. + +## Events per run + +| Event | When | Notes | +|---|---|---| +| `policy.decision` | Run start | `policy_id: toolset`, the tools offered to the model, digested. Every run has it, so a run with no tool calls still carries policy evidence and can be signed | +| `policy.decision` | Each tool call | The caller's verdict, `reason_codes` on a deny, gate time as `evaluation_duration_ns`, input digest | +| `action.executed` | Each tool call | Bound to its decision by `policy_event_id`; `outcome` success, error or denied | +| `usage.recorded` | Each model round, plus an `agent_run` rollup | Tokens including cache reads and writes; cost only when you pass `prices` | +| `data_flow.observed` | Each successful tool call | Read or write between the agent and the catalog's system, classified by the catalog | +| `data_flow.observed` | Run end | Agent → `model-provider` `anthropic`, `policy_decision: not_evaluated` | + +## What it does NOT claim + +See rules 2 and 5 in [CONTRIBUTING.md](../../CONTRIBUTING.md). + +- **It evaluates no policy.** Every decision comes from the application's own + gate, passed in by the caller. `enforcement_mode` must be stated: `enforce` only + when a deny actually blocks the call, `monitor` when the gate only observes, + `disabled` when there is no gate (TRACE records `declared`). + `policy_bundle_digest` names the gate code you say made the decisions; nothing + here checks that it did. +- **The record is only as trustworthy as the spool.** `origin.kind` is + `log-import` because the recorder reads a file the application wrote. Anything + that can write that file can write rows. Protect it like an audit log. +- **`runtime.platform` is `software-only`.** Level 0 passes. Level 1 fails on + `TR-RTE-001` and `TR-RTE-004`; nothing here carries a hardware attestation or a + verifier nonce. +- **Classification is yours.** The catalog is supplied by the caller, never inferred + from tool names. A tool missing from the catalog is treated as a write to an + `uncatalogued` system at the highest class, so a gap overstates rather than + understates. What is sent to the model is classed as the highest of + `prompt_classification` and every class a tool read; nothing evaluated that flow, + so it is marked `not_evaluated`. +- **A deny makes the appraisal `contraindicated`.** That is the upstream + `finalize_trace` rule. For a run where the gate refused a call it is the expected + result: the record shows a refused action, not a failed gate. +- **Unfinished runs are not signed.** A run with no `run.end` (crash, restart, or + idle longer than `idle_seconds`) is sealed `incomplete` and stored with + `trace: null`. Rows arriving after a run was sealed are ignored. +- **Metadata only, with one caveat.** No prompt, reply, tool argument or tool + result is written; a test puts a secret in tool inputs and results and checks it + appears nowhere in the spool, evidence, records or spans. Reason codes and policy + ids must be short tokens, otherwise they are replaced with `UNSPECIFIED`. The + input digest is an unsalted sha256, so an input with few possible values (a short + id) can be confirmed by guessing. +- **The model id is the one requested** in `run.start`. Per-round `usage.recorded` + events carry the model each response reported. + +## Run it + +Offline, no API key, against released packages: + +```bash +pip install -e "integrations/anthropic-tool-loop[recorder,test]" +python integrations/anthropic-tool-loop/examples/offline_tool_loop.py --out record.json +trace-tests verify --record record.json --level 0 +``` + +The example drives two turns of a purchasing conversation through a scripted +stand-in for `anthropic.Anthropic()`: in the first, a requester without purchasing +rights asks for stock and a purchase order, the stock lookup runs and the gate +denies the order; in the second, a buyer confirms and the order is created. Output: + +``` +turn 1: contraindicated signed=True tool_calls=2 denies=1 writes=0 data_class=internal events=10 verify=True + inventory_lookup - allow success read + purchase_orders create deny denied write WRITERS_ONLY +turn 2: affirming signed=True tool_calls=1 denies=0 writes=1 data_class=confidential events=8 verify=True + purchase_orders create allow success write + +tampered copy (data_class -> public): verify ok=False (InvalidSignature) +supplier name found in spool/records: nowhere +``` + +## What is verified + +```bash +pytest integrations/anthropic-tool-loop/tests -q +pip install anthropic==1.14.0 +pytest integrations/anthropic-tool-loop/tests/test_anthropic_interop.py -q +``` + +- `test_anthropic_spool.py` (standard library and pytest only, runs on 3.9): + rows carry no content, the action verb is opt-in and token-shaped, reason codes + are never free text, outcomes, `end` written once, dict-shaped responses, the + writer survives an unwritable spool, malformed responses and a broken lock, and + the catalog's prefix matching and conservative default. +- `test_anthropic_recorder.py`: a clean run is signed `affirming` and verifies; a + deny is `contraindicated` with its reason code and produces no write flow; an + allowed write is a confidential write flow; a chat-only run is signed; unfinished + and idle runs are unsigned; edits to appraisal, data class, enforcement mode or + subject, and a different key, all fail verification; the metadata-only check end + to end; restart mid-run; half-written lines; late rows; a malformed run is logged and + does not stop the others; no OpenTelemetry; required configuration; `trace-tests verify --level 0` on a signed record. +- `test_anthropic_interop.py`: the example's loop driven by the released + `anthropic` SDK with only its HTTP transport replaced, so a renamed field on + `Message`, `Usage` or `ToolUseBlock` fails a specific assertion. Run against + `anthropic` 0.69.0, 1.12.1, 1.13.0 and 1.14.0. + +Tested with `agentrust-telemetry` 0.1.0a6, `agentrust-trace` 0.11.0, +`agentrust-trace-tests` 0.6.2 and `opentelemetry-sdk` 1.45.1 on Python 3.11, 3.12 +and 3.13; the writer tests also on 3.9. + +## Conformance CI + +The repository-root workflow +[`.github/workflows/anthropic-tool-loop-conformance.yml`](../../.github/workflows/anthropic-tool-loop-conformance.yml) +(path-scoped to this directory) runs the writer tests on Python 3.9 with nothing +else installed, then on 3.11 to 3.13 installs the released packages, runs the +tests and the interop test, emits a record with the offline example, and runs +`trace-tests verify --level 0`. diff --git a/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/__init__.py b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/__init__.py new file mode 100644 index 0000000..8fb75c0 --- /dev/null +++ b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/__init__.py @@ -0,0 +1,11 @@ +"""Record an Anthropic Messages API tool loop as AgenTrust Telemetry events and a signed TRACE record. + +`Run`, `digest`, `ToolCatalog` and `ToolInfo` are standard library only. The recorder lives in +`anthropic_tool_loop_trace.recorder` and needs agentrust-telemetry and agentrust-trace. +""" + +from .catalog import ToolCatalog, ToolInfo +from .spool import Run, default_spool_path, digest + +__version__ = "0.1.0" +__all__ = ["Run", "digest", "default_spool_path", "ToolCatalog", "ToolInfo", "__version__"] diff --git a/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/catalog.py b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/catalog.py new file mode 100644 index 0000000..6c37cf4 --- /dev/null +++ b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/catalog.py @@ -0,0 +1,89 @@ +"""Caller-supplied tool catalog: which system a tool touches, how sensitive that data is, and +whether a call writes. + +The recorder never guesses this from tool names. You describe your own tools: + + catalog = ToolCatalog( + { + "inventory_lookup": ToolInfo("inventory-db", "database", "internal"), + "purchase_orders": ToolInfo("erp", "erp", "confidential", + write=lambda action: action not in {"get", "list"}), + "crm_*": ToolInfo("crm", "crm", "confidential"), # prefix match + }, + taxonomy="example.v1", + order=("public", "internal", "confidential"), # lowest to highest + ) + +Standard library only. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Mapping, Optional, Sequence, Union + +__all__ = ["ToolInfo", "ToolCatalog"] + +WriteRule = Union[bool, Callable[[Optional[str]], bool]] + + +@dataclass(frozen=True) +class ToolInfo: + """system: id of the system the tool reaches (becomes the data-flow endpoint id). + kind: endpoint kind, e.g. "database", "erp", "saas", "file". + classification: one value from the catalog's order. + write: True, False, or a function of the call's action verb (None when no verb was recorded). + """ + + system: str + kind: str = "service" + classification: str = "internal" + write: WriteRule = False + + +class ToolCatalog: + """Maps tool names to ToolInfo. Exact names win over "prefix*" keys; the longest prefix wins. + + A tool that is not in the catalog gets `default`. When no default is given, an unknown tool is + treated as a write to an "uncatalogued" system at the highest classification, so a missing + entry overstates rather than understates what happened. + """ + + def __init__(self, tools: Mapping[str, ToolInfo], *, taxonomy: str, order: Sequence[str], + default: Optional[ToolInfo] = None): + if not taxonomy: + raise ValueError("taxonomy is required") + order = tuple(order) + if not order or len(set(order)) != len(order): + raise ValueError("order must list each classification once, lowest first") + self.taxonomy = taxonomy + self.order = order + self.default = default or ToolInfo("uncatalogued", "other", order[-1], True) + self._exact = {} + self._prefixes = [] + for name, info in dict(tools).items(): + if info.classification not in order: + raise ValueError(f"{name}: classification {info.classification!r} is not in order") + if name.endswith("*"): + self._prefixes.append((name[:-1], info)) + else: + self._exact[name] = info + if self.default.classification not in order: + raise ValueError("default classification is not in order") + self._prefixes.sort(key=lambda item: len(item[0]), reverse=True) + + def lookup(self, tool: str) -> ToolInfo: + if tool in self._exact: + return self._exact[tool] + for prefix, info in self._prefixes: + if tool.startswith(prefix): + return info + return self.default + + def is_write(self, tool: str, action: Optional[str] = None) -> bool: + rule = self.lookup(tool).write + return bool(rule(action)) if callable(rule) else bool(rule) + + def highest(self, classes) -> str: + ranked = [c for c in classes if c in self.order] + return max(ranked, key=self.order.index) if ranked else self.order[0] diff --git a/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/recorder.py b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/recorder.py new file mode 100644 index 0000000..6338157 --- /dev/null +++ b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/recorder.py @@ -0,0 +1,451 @@ +"""Turn spool rows into AgenTrust Telemetry events, optional OpenTelemetry spans, and one signed +TRACE record per finished run. + +Runs out of process (a sidecar, a cron job, or the same host on a timer) and needs Python 3.11+, +agentrust-telemetry and agentrust-trace. OpenTelemetry is optional: pass a TracerProvider to get +spans with the real start and end times of each model round and tool call. + +Per finished run it emits, in order of the observed timeline: + + policy.decision one run-level decision for the toolset offered to the model, then one per + tool call carrying the caller's gate verdict and reason code + action.executed one per tool call, bound to its policy decision + usage.recorded one per model round, plus an agent_run rollup + data_flow.observed one per successful tool call (read or write, classified by the caller's + catalog), plus one agent -> model-provider flow marked not_evaluated, + because nothing evaluated what was sent to the model + +The events are chained by an EvidenceAccumulator and sealed. A complete run is finalized into a +TRACE v0.2 record (origin.kind "log-import") and signed. A run that never wrote run.end is sealed +"incomplete" and is not signed. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import threading +import logging +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Literal + +from agentrust_telemetry import ( + EventFactory, + EvidenceAccumulator, + SchemaValidator, + TelemetryClient, + TraceConfiguration, + UsageAccumulator, + finalize_trace, +) +from agentrust_trace import jwk_thumbprint, key_to_jwk, verify_record + +from . import __version__ +from .catalog import ToolCatalog + +try: # optional: spans are produced only when a TracerProvider is passed + from opentelemetry import trace as _otel +except ImportError: # pragma: no cover - exercised only without the SDK installed + _otel = None + +__all__ = ["Recorder", "verify", "source_digest", "PRODUCER"] + +PRODUCER = "anthropic-tool-loop-trace" +RUN_ID_RE = re.compile(r"^run-[0-9a-f]{32}$") +DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_KINDS = {"run.start", "model.call", "tool.call", "run.end"} +_log = logging.getLogger("anthropic_tool_loop_trace.recorder") + + +def source_digest(root: str | os.PathLike, pattern: str = "*.py") -> str: + """sha256 over every file matching `pattern` under `root` (relative path + bytes, sorted). + + Use it for build_digest (your agent's code) or policy_bundle_digest (your gate code).""" + root = Path(root) + h = hashlib.sha256() + for path in sorted(p for p in root.rglob(pattern) if "__pycache__" not in p.parts): + h.update(path.relative_to(root).as_posix().encode() + b"\0" + path.read_bytes() + b"\0") + return "sha256:" + h.hexdigest() + + +def verify(record: dict, public_jwk: dict) -> dict: + """Check a stored TRACE record's signature and structure. Never raises.""" + try: + result = verify_record(record, public_jwk, max_age_seconds=None) + return {"ok": True, "profile": getattr(result, "profile", None)} + except Exception as exc: + return {"ok": False, "error": f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__} + + +def _day(ns: int) -> str: + return datetime.fromtimestamp(ns / 1e9, tz=timezone.utc).strftime("%Y-%m-%d") + + +class _EvidenceStore: + """Append-only JSONL per UTC day, fsynced, idempotent by event id.""" + + def __init__(self, root: Path): + self.root = root + self.root.mkdir(parents=True, exist_ok=True) + self.seen: set[str] = set() + + def append(self, entry) -> bool: + if entry.event_id in self.seen: + return True + path = self.root / f"{_day(int(entry.event['time_unix_nano']))}.jsonl" + row = {"run_id": entry.event["run_id"], "sequence": entry.sequence, "event_id": entry.event_id, + "previous_digest": entry.previous_digest, "digest": entry.digest, "event": entry.event} + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") + fh.flush() + os.fsync(fh.fileno()) + self.seen.add(entry.event_id) + return True + + +@dataclass +class Recorder: + """Reads the spool, records finished runs, keeps its read position across restarts. + + spool, data_dir: where the agent writes rows; where records, evidence and state go. + signing_key: Ed25519 private key (agentrust_trace.load_key / generate_key). + catalog: your ToolCatalog. + enforcement_mode: "enforce" if your gates block denied calls, "monitor" if they only + observe, "disabled" if there is no gate (TRACE records "declared"). + Required: there is no safe default. + policy_bundle_digest: "sha256:" of the gate code or policy that made the decisions. + build_digest: "sha256:" of the agent build that produced the run. + prompt_classification: what class of data your prompts carry. The recorder cannot see them, + so you state it; the model-provider flow is at least this class. + appraisal_verifier: a URI naming who appraised the record (e.g. your ledger's URL). + Defaults to urn:anthropic-tool-loop-trace:. + """ + + spool: str | os.PathLike + data_dir: str | os.PathLike + signing_key: Any + catalog: ToolCatalog + enforcement_mode: Literal["enforce", "monitor", "disabled"] + policy_bundle_digest: str + build_digest: str + prompt_classification: str + policy_engine: str = "application-gates" + producer_instance_id: str | None = None + appraisal_verifier: str | None = None + tracer_provider: Any = None + idle_seconds: int = 1200 + prices: dict[str, tuple[float, float, float, float]] | None = None + pricing_version: str | None = None + + def __post_init__(self): + if self.enforcement_mode not in ("enforce", "monitor", "disabled"): + raise ValueError("enforcement_mode must be enforce, monitor or disabled") + for name in ("policy_bundle_digest", "build_digest"): + if not DIGEST_RE.match(getattr(self, name) or ""): + raise ValueError(f"{name} must look like sha256:<64 hex>") + if self.prompt_classification not in self.catalog.order: + raise ValueError("prompt_classification is not in the catalog's order") + if self.prices and not self.pricing_version: + raise ValueError("pricing_version is required with prices") + self.spool = Path(self.spool).expanduser() + self.data_dir = Path(self.data_dir).expanduser() + (self.data_dir / "records").mkdir(parents=True, exist_ok=True) + self._state_path = self.data_dir / "state.json" + self._index_path = self.data_dir / "index.jsonl" + self._validator = SchemaValidator.bundled() + self._factory = EventFactory(self._validator, producer_name=PRODUCER, producer_version=__version__, + producer_instance_id=self.producer_instance_id) + self._evidence = _EvidenceStore(self.data_dir / "evidence") + self.public_jwk = {k: v for k, v in key_to_jwk(self.signing_key).items() if k != "d"} + self.key_thumbprint = jwk_thumbprint(self.public_jwk) + self._tracer = None + if self.tracer_provider is not None: + if _otel is None: + raise ValueError("tracer_provider given but opentelemetry-api is not installed") + self._tracer = self.tracer_provider.get_tracer(PRODUCER, __version__) + self._lock = threading.Lock() + saved = json.loads(self._state_path.read_text()) if self._state_path.exists() else {} + self.offset = int(saved.get("offset", 0)) + self.pending: dict[str, list[dict]] = saved.get("pending", {}) + + # -- spool ----------------------------------------------------------------------------- + def _save_state(self): + tmp = self._state_path.with_suffix(".tmp") + tmp.write_text(json.dumps({"offset": self.offset, "pending": self.pending})) + os.replace(tmp, self._state_path) + + def poll(self, final: bool = False) -> list[str]: + """Read new spool rows and record every run that ended, or went idle, or (final=True) + everything still pending. Returns the run ids recorded.""" + done: list[str] = [] + if self.spool.exists(): + if self.spool.stat().st_size < self.offset: # rotated or truncated + self.offset = 0 + with self.spool.open("rb") as fh: + fh.seek(self.offset) + chunk = fh.read() + usable = chunk[: chunk.rfind(b"\n") + 1] # never consume a half-written line + self.offset += len(usable) + sealed: dict[str, bool] = {} + for raw in usable.splitlines(): + try: + row = json.loads(raw) + except ValueError: + continue + run_id = row.get("run_id") if isinstance(row, dict) else None + if not isinstance(run_id, str) or not RUN_ID_RE.match(run_id) or row.get("kind") not in _KINDS: + continue + if run_id not in self.pending: + if run_id not in sealed: + sealed[run_id] = self._record_path(run_id) is not None + if sealed[run_id]: + continue # already recorded; a late row cannot reopen a record + self.pending.setdefault(run_id, []).append(row) + now = time.time_ns() + for run_id, rows in list(self.pending.items()): + ended = any(r["kind"] == "run.end" for r in rows) + last = max(int(r.get("end") or r.get("t") or 0) for r in rows) + if ended or final or now - last > self.idle_seconds * 1_000_000_000: + try: + self.record(rows, complete=ended) + except Exception: + _log.exception("recording %s failed; its rows are dropped", run_id) + done.append(run_id) + del self.pending[run_id] + self._save_state() + return done + + # -- reads ----------------------------------------------------------------------------- + def _record_path(self, run_id: str) -> Path | None: + return next(iter(sorted((self.data_dir / "records").glob(f"*/{run_id}.json"))), None) + + def run(self, run_id: str) -> dict | None: + """The stored {"run", "events", "trace"} for a run id, or None.""" + if not RUN_ID_RE.match(run_id or ""): + return None + path = self._record_path(run_id) + return json.loads(path.read_text()) if path else None + + def verify(self, record: dict) -> dict: + return verify(record, self.public_jwk) + + # -- recording ------------------------------------------------------------------------- + def _cost(self, row: dict) -> float | None: + price = (self.prices or {}).get(row["model"]) + if not price: + return None + p_in, p_out, p_read, p_write = price + return round((row["input"] * p_in + row["output"] * p_out + row["cache_read"] * p_read + + row["cache_write"] * p_write) / 1_000_000, 6) + + def record(self, rows: list[dict], complete: bool) -> dict: + with self._lock: + return self._record(rows, complete) + + def _record(self, rows: list[dict], complete: bool) -> dict: + rows = sorted(rows, key=lambda r: int(r.get("start") or r.get("t") or 0)) + start = next((r for r in rows if r["kind"] == "run.start"), None) + end = next((r for r in rows if r["kind"] == "run.end"), None) + models = [r for r in rows if r["kind"] == "model.call"] + tools = [r for r in rows if r["kind"] == "tool.call"] + complete = bool(complete and start and end) + run_id = rows[0]["run_id"] + agent_id = (start or {}).get("agent_id") or "unknown-agent" + agent_name = (start or {}).get("agent_name") or "unknown" + model = (start or {}).get("model") or (models[0]["model"] if models else "unknown") + workflow_id = (start or {}).get("workflow_id") + t_start = int((start or {}).get("t") or min(int(r.get("start") or r.get("t")) for r in rows)) + t_end = int((end or {}).get("t") or max(int(r.get("end") or r.get("t") or t_start) for r in rows)) + common = {"run_id": run_id, "agent_id": agent_id} + if workflow_id: + common["workflow_id"] = workflow_id + cat = self.catalog + bundle = {"algorithm": "sha256", "value": self.policy_bundle_digest[7:]} + policy_base = {"engine": self.policy_engine, "engine_version": self.policy_bundle_digest[7:19], + "bundle_digest": bundle} + + events: list[dict] = [] + current = {"span": None} + accumulator = EvidenceAccumulator(run_id, self._validator, durable_append=self._evidence.append) + client = TelemetryClient(self._validator, span_resolver=lambda: current["span"], + evidence_sink=accumulator) + usage = UsageAccumulator(self._validator) + otel = _otel + def ids(span): + if span is None: + return {} + c = span.get_span_context() + return {"trace_id": f"{c.trace_id:032x}", "span_id": f"{c.span_id:016x}"} + + def emit(span, event_type, at_ns, **payload): + event = self._factory.build(event_type, **common, **ids(span), time_unix_nano=int(at_ns), **payload) + current["span"] = span + client.emit(event) + events.append(event) + return event + + def start_span(name, at_ns, attributes, parent=None): + if self._tracer is None: + return None + return self._tracer.start_span(name, context=parent, start_time=int(at_ns), attributes=attributes) + + root = start_span(f"invoke_agent {agent_name}", t_start, { + "gen_ai.operation.name": "invoke_agent", "gen_ai.agent.name": agent_name, + "gen_ai.agent.id": agent_id, "gen_ai.provider.name": "anthropic", + "gen_ai.request.model": model, "agentrust.run_id": run_id}) + ctx = otel.set_span_in_context(root) if root is not None else None + + # Run-level decision: the toolset offered to the model. It is the policy decision every run + # has, so a run with no tool calls still carries policy evidence and can be signed. + toolset = (start or {}).get("toolset") or [] + emit(root, "policy.decision", t_start, decision="allow", action_type="agent.invoke", + resource_type="toolset", enforcement_mode=self.enforcement_mode, evaluation_duration_ns=0, + policy={**policy_base, "policy_id": "toolset", + "input_digest": {"algorithm": "sha256", + "value": hashlib.sha256(json.dumps(sorted(toolset)).encode()).hexdigest()}}) + + summary_tools, read_classes = [], [] + tokens = {"input": 0, "output": 0, "cache_read": 0, "cache_write": 0} + cost_total, priced = 0.0, True + for row in sorted(models + tools, key=lambda r: int(r["start"])): + if row["kind"] == "model.call": + span = start_span(f"chat {row['model']}", row["start"], { + "gen_ai.operation.name": "chat", "gen_ai.provider.name": "anthropic", + "gen_ai.request.model": row["model"], "gen_ai.usage.input_tokens": row["input"], + "gen_ai.usage.output_tokens": row["output"], + "gen_ai.usage.cache_read.input_tokens": row["cache_read"], + "gen_ai.usage.cache_creation.input_tokens": row["cache_write"], + "gen_ai.response.finish_reasons": [str(row.get("stop_reason") or "")], + "agentrust.round": row["round"]}, ctx) + payload = {"scope": "model_call", "operation": "chat", "provider": "anthropic", + "request_model": row["model"] or model, "input_tokens": row["input"], + "output_tokens": row["output"], "cache_read_tokens": row["cache_read"], + "cache_write_tokens": row["cache_write"]} + amount = self._cost(row) + if amount is not None: + payload["cost"] = {"amount": amount, "currency": "USD", "source": "estimate", + "pricing_version": self.pricing_version} + cost_total += amount + else: + priced = False + usage.add(emit(span, "usage.recorded", row["end"], **payload)) + for k in tokens: + tokens[k] += row[k] + if span is not None: + span.end(end_time=int(row["end"])) + continue + + tool = row["tool"] + info = cat.lookup(tool) + write = cat.is_write(tool, row.get("action")) + span = start_span(f"execute_tool {tool}", row["start"], { + "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": tool, + "gen_ai.tool.call.id": row["tool_use_id"], "agentrust.policy": row["policy"], + "agentrust.decision": row["decision"], "agentrust.outcome": row["outcome"], + "agentrust.system": info.system, "agentrust.write": write, "agentrust.round": row["round"]}, ctx) + gate_ns = int(row.get("gate_ns") or 0) + policy = emit(span, "policy.decision", int(row["start"]) + gate_ns, decision=row["decision"], + action_type=f"{info.system}.{'write' if write else 'read'}", + resource_type=info.system, enforcement_mode=self.enforcement_mode, + evaluation_duration_ns=gate_ns, + policy={**policy_base, "policy_id": row["policy"], + "input_digest": {"algorithm": "sha256", "value": row["input_digest"]}}, + **({"reason_codes": [row["reason"]]} if row.get("reason") else {})) + action = {"action_id": row["tool_use_id"], "action_kind": "tool", "action_name": tool, + "operation": row.get("action") or ("write" if write else "read"), + "outcome": row["outcome"], "duration_ns": max(0, int(row["end"]) - int(row["start"])), + "action_digest": {"algorithm": "sha256", "value": row["input_digest"]}, + "target": {"kind": info.kind, "id": info.system}, "policy_event_id": policy["event_id"]} + if row.get("error_type"): + action["error_type"] = row["error_type"] + emit(span, "action.executed", row["end"], **action) + if row["outcome"] == "success" and not row.get("tool_error"): + me, there = {"kind": "agent", "id": agent_id}, {"kind": info.kind, "id": info.system} + emit(span, "data_flow.observed", row["end"], direction="write" if write else "read", + source=me if write else there, destination=there if write else me, + classification={"taxonomy": cat.taxonomy, "value": info.classification, + "producer": f"{PRODUCER} tool catalog"}, + purpose=f"tool call ({agent_name})", policy_decision=row["decision"], transformation="none") + if not write: + read_classes.append(info.classification) + if span is not None: + if row["outcome"] != "success": + span.set_status(otel.Status(otel.StatusCode.ERROR, + f"{row['outcome']}: {row.get('reason') or row.get('error_type') or ''}")) + span.end(end_time=int(row["end"])) + summary_tools.append({"tool": tool, "action": row.get("action"), "system": info.system, + "write": write, "classification": info.classification, + "outcome": row["outcome"], "decision": row["decision"], "policy": row["policy"], + "reason": row.get("reason"), "error_type": row.get("error_type"), + "tool_error": bool(row.get("tool_error")), "round": row["round"], + "start": row["start"], "ms": round((int(row["end"]) - int(row["start"])) / 1e6, 1)}) + + # What went to the model provider: the prompt (classified by the caller) plus everything the + # tools read, which the loop hands back to the model. Nothing evaluated that flow. + sent_class = cat.highest(read_classes + [self.prompt_classification]) + if models: + emit(root, "data_flow.observed", t_end, direction="send", + source={"kind": "agent", "id": agent_id}, destination={"kind": "model-provider", "id": "anthropic"}, + classification={"taxonomy": cat.taxonomy, "value": sent_class, + "producer": f"{PRODUCER} tool catalog"}, + purpose="inference", policy_decision="not_evaluated", transformation="none") + rollup = usage.rollup(self._factory, scope="agent_run", run_id=run_id, operation="invoke_agent", + agent_id=agent_id, workflow_id=workflow_id) + rollup = {**rollup, **common, **ids(root), "time_unix_nano": str(t_end)} + self._validator.validate(rollup) + current["span"] = root + client.emit(rollup) + events.append(rollup) + + if root is not None: + root.set_attribute("agentrust.complete", complete) + root.set_attribute("agentrust.tool_calls", len(tools)) + if any(t["outcome"] == "denied" for t in summary_tools): + root.set_status(otel.Status(otel.StatusCode.ERROR, "a gate denied a tool call")) + root.end(end_time=t_end) + + snapshot = accumulator.seal(completeness="complete" if complete else "incomplete") + record, trace_error = None, None + if complete: + try: + record = finalize_trace(snapshot, TraceConfiguration( + subject=agent_id, model_provider="anthropic", model_id=model, build_digest=self.build_digest, + origin_kind="log-import", origin_producer=f"{PRODUCER}/{__version__}", + appraisal_verifier=self.appraisal_verifier + or f"urn:{PRODUCER}:{self.producer_instance_id or 'recorder'}", + classification_taxonomy=cat.taxonomy, classification_order=cat.order), + signing_key=self.signing_key) + except Exception as exc: + trace_error = f"{type(exc).__name__}: {exc}" + else: + trace_error = "run never finished (no run.end): evidence sealed incomplete, not signed" + + summary = { + "run_id": run_id, "agent_id": agent_id, "agent_name": agent_name, "workflow_id": workflow_id, + "model": model, "start": t_start, "end": t_end, "duration_ms": round((t_end - t_start) / 1e6), + "stop": (end or {}).get("stop") or "unfinished", "rounds": len(models), "tool_calls": len(tools), + "denies": sum(t["outcome"] == "denied" for t in summary_tools), + "errors": sum(t["outcome"] == "error" or t["tool_error"] for t in summary_tools), + "writes": sum(t["write"] and t["outcome"] == "success" for t in summary_tools), + "systems": sorted({t["system"] for t in summary_tools}), "tokens": tokens, + "toolset_size": len(toolset), + "cost_usd": round(cost_total, 6) if priced and models and self.prices else None, + "data_class": record["data_class"] if record else sent_class, + "sent_to_model": sent_class if models else None, + "verdict": record["appraisal"]["status"] if record else "unsigned", + "signed": record is not None, "complete": complete, "trace_error": trace_error, + "otel_trace_id": ids(root).get("trace_id"), "events": len(events), + "evidence_chain": snapshot.chain_digest, "build": self.build_digest, "tools": summary_tools, + } + out = self.data_dir / "records" / _day(t_start) + out.mkdir(parents=True, exist_ok=True) + (out / f"{run_id}.json").write_text(json.dumps( + {"run": summary, "events": events, "trace": record}, indent=1, default=str)) + with self._index_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps({k: v for k, v in summary.items() if k != "tools"}, default=str) + "\n") + return summary diff --git a/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/spool.py b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/spool.py new file mode 100644 index 0000000..4ca5d67 --- /dev/null +++ b/integrations/anthropic-tool-loop/anthropic_tool_loop_trace/spool.py @@ -0,0 +1,176 @@ +"""In-process spool writer for an Anthropic Messages API tool loop. + +Standard library only and importable on Python 3.9, so it can sit inside an +existing agent without adding dependencies. Every method is fail-open: a spool +problem logs one warning and never raises into the agent's reply path. + +Rows are metadata only. No prompt, reply, tool argument or tool result is +written. A row carries names, counts, timings, the caller's gate decision, and a +sha256 digest of each tool input, so a later audit can show which exact call ran +without the call's content leaving the process. + +Row kinds, one JSON object per line: + + run.start agent id/name, requested model, toolset, optional workflow id + model.call one per messages.create round: model, token usage, stop reason + tool.call one per tool_use block: decision, outcome, input digest, timings + run.end why the loop stopped +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import re +import threading +import time +import uuid + +__all__ = ["Run", "digest", "default_spool_path", "SPOOL_ENV"] + +SPOOL_ENV = "AGENTRUST_ANTHROPIC_SPOOL" +_LOCK = threading.Lock() +_VERB_RE = re.compile(r"^[a-z][a-z0-9_]{0,39}$") +_CODE_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,64}$") +_log = logging.getLogger("anthropic_tool_loop_trace") + + +def default_spool_path() -> str: + """$AGENTRUST_ANTHROPIC_SPOOL, else ~/.local/state/agentrust/anthropic-spool.jsonl.""" + return os.environ.get(SPOOL_ENV) or os.path.expanduser( + "~/.local/state/agentrust/anthropic-spool.jsonl" + ) + + +def digest(value) -> str: + """sha256 over canonical JSON (sorted keys, no whitespace). Same input, same digest.""" + try: + blob = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + except Exception: + blob = repr(value) + return hashlib.sha256(blob.encode("utf-8")).hexdigest() + + +def _get(obj, name, default=None): + """Read a field from an SDK object or a plain dict.""" + if isinstance(obj, dict): + return obj.get(name, default) + return getattr(obj, name, default) + + +def _code(value): + """A reason code or policy id: short and token-shaped, never free text.""" + if value is None: + return None + value = str(value) + return value if _CODE_RE.match(value) else "UNSPECIFIED" + + +class Run: + """One pass through the tool loop (one user turn). Every method is fail-open. + + agent_id: stable identity of the agent, e.g. a SPIFFE id. Becomes the TRACE subject. + agent_name: short display name. + model: the model requested in messages.create. + toolset: names of the tools offered to the model on this run. + workflow_id: optional opaque id grouping runs (a ticket, a thread). Do not put content here. + action_field: optional name of a tool-input field holding a verb such as "get" or "create". + When set, its value is recorded if it is a short lowercase token; nothing else + from the input is. + spool: path of the JSONL spool; defaults to default_spool_path(). + """ + + def __init__(self, agent_id, agent_name, model, toolset=None, workflow_id=None, + action_field=None, spool=None): + self.id = "run-" + uuid.uuid4().hex + self.ended = False + self._action_field = action_field + self._spool = spool or default_spool_path() + try: + names = sorted({str(t) for t in toolset}) if toolset else [] + self._write({"kind": "run.start", "t": time.time_ns(), "agent_id": str(agent_id), + "agent_name": str(agent_name), "model": str(model), "toolset": names, + "workflow_id": str(workflow_id) if workflow_id else None}) + except Exception as exc: + _log.warning("agentrust spool: run.start failed: %s", exc) + + # -- writing --------------------------------------------------------------------------- + def _write(self, row): + try: + line = json.dumps({"run_id": self.id, **row}, separators=(",", ":"), default=str) + "\n" + with _LOCK: + parent = os.path.dirname(self._spool) + if parent and not os.path.isdir(parent): + os.makedirs(parent, exist_ok=True) + # One write() per row in append mode: a reader sees whole lines or a trailing + # partial line it must leave for the next poll. + with open(self._spool, "a", encoding="utf-8") as fh: + fh.write(line) + except Exception as exc: + _log.warning("agentrust spool: write failed: %s", exc) + + def _verb(self, tool_input): + if not self._action_field or not isinstance(tool_input, dict): + return None + value = tool_input.get(self._action_field) + return value if isinstance(value, str) and _VERB_RE.match(value) else None + + # -- hooks ----------------------------------------------------------------------------- + def model_call(self, round_no, start_ns, response): + """After each client.messages.create. `response` is the Message (or a dict shaped like one).""" + try: + usage = _get(response, "usage") or {} + self._write({"kind": "model.call", "round": int(round_no), + "model": str(_get(response, "model") or ""), "start": int(start_ns), + "end": time.time_ns(), "stop_reason": _get(response, "stop_reason"), + "input": int(_get(usage, "input_tokens", 0) or 0), + "output": int(_get(usage, "output_tokens", 0) or 0), + "cache_read": int(_get(usage, "cache_read_input_tokens", 0) or 0), + "cache_write": int(_get(usage, "cache_creation_input_tokens", 0) or 0)}) + except Exception as exc: + _log.warning("agentrust spool: model_call failed: %s", exc) + + def tool_call(self, round_no, block, start_ns, decision="allow", policy="toolset", reason=None, + outcome=None, error_type=None, result=None, gate_ns=0): + """After each tool_use block is handled. + + block: the tool_use content block (name, id, input). + decision: your gate's verdict, "allow" or "deny". + policy: which check decided, e.g. "toolset" or "writers_only". Token-shaped. + reason: a short reason code for a deny, e.g. "WRITERS_ONLY". Never the message text. + outcome: "success" | "error" | "denied". Defaults to "denied" for a deny, "error" when + error_type is given, else "success". + error_type: exception class name when the tool raised. + result: the tool's return value. Only its length, and whether it is a dict with an + "error" key, are recorded. + gate_ns: how long the gate check took, in nanoseconds. + """ + try: + tool_input = _get(block, "input") + if decision not in ("allow", "deny"): + decision = "deny" + if outcome not in ("success", "error", "denied"): + outcome = "denied" if decision == "deny" else ("error" if error_type else "success") + self._write({"kind": "tool.call", "round": int(round_no), "tool": str(_get(block, "name")), + "tool_use_id": str(_get(block, "id")), "action": self._verb(tool_input), + "input_digest": digest(tool_input), "start": int(start_ns), + "end": time.time_ns(), "outcome": outcome, "policy": _code(policy) or "toolset", + "decision": decision, "reason": _code(reason), + "error_type": _code(error_type), "gate_ns": int(gate_ns or 0), + "result_chars": len(str(result)) if result is not None else None, + "tool_error": bool(isinstance(result, dict) and "error" in result + and outcome == "success")}) + except Exception as exc: + _log.warning("agentrust spool: tool_call failed: %s", exc) + + def end(self, stop): + """When the loop stops: "end_turn", "max_tokens", "max_rounds", or your own token.""" + if self.ended: + return + self.ended = True + try: + self._write({"kind": "run.end", "t": time.time_ns(), "stop": _code(stop)}) + except Exception as exc: + _log.warning("agentrust spool: run.end failed: %s", exc) diff --git a/integrations/anthropic-tool-loop/examples/offline_tool_loop.py b/integrations/anthropic-tool-loop/examples/offline_tool_loop.py new file mode 100644 index 0000000..1f5af7a --- /dev/null +++ b/integrations/anthropic-tool-loop/examples/offline_tool_loop.py @@ -0,0 +1,224 @@ +"""Offline end-to-end run: no network, no API key. + +A scripted stand-in for `anthropic.Anthropic()` drives a two-turn purchasing conversation through +an ordinary Messages API tool loop: + + turn 1 a requester without purchasing rights asks for stock and a purchase order. + The stock lookup runs; the gate denies the purchase order (WRITERS_ONLY). + turn 2 a buyer confirms. The purchase order is created (an allowed write). + +The recorder then turns the spool into AgenTrust events, signs one TRACE record per turn, verifies +both, and shows that an edited copy fails verification. + + python examples/offline_tool_loop.py [--out record.json] [--keep DIR] + +--out writes the signed record of turn 2 for `trace-tests verify --record record.json --level 0`. +""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import inspect +import json +import pathlib +import sys +import tempfile +import time +from types import SimpleNamespace + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from agentrust_trace import generate_key # noqa: E402 + +from anthropic_tool_loop_trace import Run, ToolCatalog, ToolInfo # noqa: E402 +from anthropic_tool_loop_trace.recorder import Recorder # noqa: E402 + +MODEL = "claude-sonnet-5-5" +AGENT_ID = "spiffe://example.org/agent/purchasing" +WRITERS = {"buyer"} +READ_VERBS = {"get", "list"} + +TOOLS = [ + {"name": "inventory_lookup", "description": "Units on hand for a SKU.", + "input_schema": {"type": "object", "properties": {"sku": {"type": "string"}}, "required": ["sku"]}}, + {"name": "purchase_orders", "description": "Get, list or create purchase orders.", + "input_schema": {"type": "object", "properties": { + "action": {"type": "string", "enum": ["get", "list", "create"]}, + "supplier": {"type": "string"}, "sku": {"type": "string"}, "qty": {"type": "integer"}}, + "required": ["action"]}}, +] + +CATALOG = ToolCatalog( + { + "inventory_lookup": ToolInfo("inventory-db", "database", "internal"), + "purchase_orders": ToolInfo("erp", "erp", "confidential", write=lambda verb: verb not in READ_VERBS), + }, + taxonomy="example.v1", + order=("public", "internal", "confidential"), +) + + +# --- the application's own tools and gate ------------------------------------------------- +def inventory_lookup(args): + return {"sku": args["sku"], "on_hand": 14} + + +def purchase_orders(args): + return {"po": "PO-0001", "status": "draft", "qty": args.get("qty")} + + +FUNCTIONS = {"inventory_lookup": inventory_lookup, "purchase_orders": purchase_orders} + + +def gate(tool, requester, args): + """The application's own rule: only listed writers may create purchase orders. + Returns a reason code for a deny, None to allow.""" + if tool == "purchase_orders" and args.get("action") not in READ_VERBS and requester not in WRITERS: + return "WRITERS_ONLY" + return None + + +# --- the tool loop; the recording lines are numbered ------------------------------------------- +def answer(client, requester, text, spool): + messages = [{"role": "user", "content": text}] + run = Run(AGENT_ID, "purchasing", MODEL, toolset=[t["name"] for t in TOOLS], # 1 + action_field="action", spool=spool) + for round_no in range(10): + t0 = time.time_ns() + response = client.messages.create(model=MODEL, max_tokens=1024, tools=TOOLS, messages=messages) + run.model_call(round_no, t0, response) # 2 + if response.stop_reason != "tool_use": + run.end(response.stop_reason) # 3 + return "".join(b.text for b in response.content if b.type == "text") + results = [] + for block in response.content: + if block.type != "tool_use": + continue + t0 = time.time_ns() + denied = gate(block.name, requester, block.input) + gate_ns = time.time_ns() - t0 + result, error = None, None + if not denied: + try: + result = FUNCTIONS[block.name](block.input) + except Exception as exc: + error = exc + run.tool_call(round_no, block, t0, decision="deny" if denied else "allow", # 4 + policy="purchasing_gate", reason=denied, result=result, + error_type=type(error).__name__ if error else None, gate_ns=gate_ns) + if denied or error: + content = "Only buyers can create purchase orders." if denied else str(error) + results.append({"type": "tool_result", "tool_use_id": block.id, "is_error": True, + "content": content}) + else: + results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result)}) + messages += [{"role": "assistant", "content": response.content}, {"role": "user", "content": results}] + run.end("max_rounds") # 5 + return "Stopped after too many rounds." + + +# --- a scripted stand-in for anthropic.Anthropic() ----------------------------------------- +def _message(stop_reason, *content, usage=(410, 60, 0, 0)): + i, o, r, w = usage + return SimpleNamespace( + id="msg_scripted", model=MODEL, role="assistant", stop_reason=stop_reason, content=list(content), + usage=SimpleNamespace(input_tokens=i, output_tokens=o, cache_read_input_tokens=r, + cache_creation_input_tokens=w)) + + +def _text(text): + return SimpleNamespace(type="text", text=text) + + +def _tool_use(tool_id, name, args): + return SimpleNamespace(type="tool_use", id=tool_id, name=name, input=args) + + +class ScriptedClient: + """Answers messages.create from a fixed script. Same call shape as the SDK, nothing leaves the process.""" + + def __init__(self, script): + self._script = list(script) + self.messages = self + self.calls = 0 + + def create(self, **kwargs): + assert kwargs["model"] and kwargs["messages"] and kwargs["tools"], "called like messages.create" + self.calls += 1 + return self._script.pop(0) + + +SUPPLIER = "Northwind Supply, account 55-0193" # appears in tool inputs; must never reach the spool + +SCRIPT = [ + # turn 1, round 0: look up stock and try to create a purchase order + _message("tool_use", _text("Checking stock and drafting the order."), + _tool_use("toolu_01", "inventory_lookup", {"sku": "SKU-118"}), + _tool_use("toolu_02", "purchase_orders", + {"action": "create", "supplier": SUPPLIER, "sku": "SKU-118", "qty": 20})), + # turn 1, round 1: the gate refused; the model explains + _message("end_turn", _text("14 on hand. A buyer has to create the purchase order."), usage=(520, 40, 400, 0)), + # turn 2, round 0: the buyer confirms + _message("tool_use", _tool_use("toolu_03", "purchase_orders", + {"action": "create", "supplier": SUPPLIER, "sku": "SKU-118", "qty": 20})), + # turn 2, round 1 + _message("end_turn", _text("Created PO-0001 as a draft."), usage=(480, 20, 400, 0)), +] + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--out", help="write the signed TRACE record of turn 2 here") + ap.add_argument("--keep", help="use this directory for spool and records instead of a temp dir") + args = ap.parse_args(argv) + + work = pathlib.Path(args.keep or tempfile.mkdtemp(prefix="anthropic-tool-loop-")) + work.mkdir(parents=True, exist_ok=True) + spool = work / "spool.jsonl" + client = ScriptedClient(SCRIPT) + + print("turn 1:", answer(client, "requester", "How many SKU-118 do we have? Order 20 more.", str(spool))) + print("turn 2:", answer(client, "buyer", "Go ahead with the 20.", str(spool))) + + recorder = Recorder( + spool=spool, data_dir=work / "data", signing_key=generate_key(), catalog=CATALOG, + enforcement_mode="enforce", + policy_bundle_digest="sha256:" + hashlib.sha256(inspect.getsource(gate).encode()).hexdigest(), + build_digest="sha256:" + hashlib.sha256(pathlib.Path(__file__).read_bytes()).hexdigest(), + prompt_classification="internal", + ) + recorded = recorder.poll() + print(f"\nrecorded {len(recorded)} run(s) in {work / 'data'}") + + stored = [recorder.run(run_id) for run_id in recorded] + for n, rec in enumerate(stored, 1): + s = rec["run"] + print(f"turn {n}: {s['verdict']:<16} signed={s['signed']} tool_calls={s['tool_calls']} " + f"denies={s['denies']} writes={s['writes']} data_class={s['data_class']} " + f"events={s['events']} verify={recorder.verify(rec['trace'])['ok'] if rec['trace'] else '-'}") + if s["trace_error"]: + print(f" not signed: {s['trace_error']}") + for t in s["tools"]: + print(f" {t['tool']:<17} {t['action'] or '-':<7} {t['decision']:<5} {t['outcome']:<8} " + f"{'write' if t['write'] else 'read':<5} {t['reason'] or ''}") + + tampered = copy.deepcopy(stored[1]["trace"]) + tampered["data_class"] = "public" + check = recorder.verify(tampered) + print(f"\ntampered copy (data_class -> public): verify ok={check['ok']} ({check.get('error', '')[:80]})") + + leaked = [p.name for p in work.rglob("*") if p.is_file() and SUPPLIER.split(",")[0] in p.read_text()] + print(f"supplier name found in spool/records: {leaked or 'nowhere'}") + + if args.out: + pathlib.Path(args.out).write_text(json.dumps(stored[1]["trace"], indent=1)) + print(f"wrote {args.out}") + ok = (stored[0]["run"]["verdict"] == "contraindicated" and stored[1]["run"]["verdict"] == "affirming" + and all(recorder.verify(r["trace"])["ok"] for r in stored) and not check["ok"] and not leaked) + return 0 if ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/integrations/anthropic-tool-loop/integration.yaml b/integrations/anthropic-tool-loop/integration.yaml new file mode 100644 index 0000000..c956730 --- /dev/null +++ b/integrations/anthropic-tool-loop/integration.yaml @@ -0,0 +1,22 @@ +name: Anthropic SDK tool loop +vendor: Disruptive Beverages +integrates_with: + - trace +description: >- + Records a Claude tool loop built on the anthropic Python SDK as AgenTrust + Telemetry events, metadata only, and signs one log-import TRACE record per run. +maintainer: + github: cosmo-lgtm +repository: https://github.com/agentrust-io/integrations +license: Apache-2.0 +tier: community +marketplace: + category: Frameworks + mark: AN + keywords: [anthropic, claude, tool-use, telemetry, log-import] +trace_roles: + - record-producer +trace_conformance_level: 0 +tested_against: + agentrust-trace: "0.11.0" + agentrust-trace-tests: "0.6.2" diff --git a/integrations/anthropic-tool-loop/pyproject.toml b/integrations/anthropic-tool-loop/pyproject.toml new file mode 100644 index 0000000..688fd39 --- /dev/null +++ b/integrations/anthropic-tool-loop/pyproject.toml @@ -0,0 +1,28 @@ +[build-system] +requires = ["setuptools>=77"] +build-backend = "setuptools.build_meta" + +[project] +name = "anthropic-tool-loop-trace" +version = "0.1.0" +description = "Records an Anthropic Messages API tool loop as AgenTrust Telemetry events and signs one TRACE record per run" +readme = "README.md" +# The in-process writer is standard library only and runs on 3.9. The recorder extra needs 3.11+. +requires-python = ">=3.9" +license = "Apache-2.0" +dependencies = [] + +[project.optional-dependencies] +recorder = [ + "agentrust-telemetry==0.1.0a6", + "agentrust-trace>=0.11.0", +] +otel = ["opentelemetry-sdk>=1.25"] +test = [ + "pytest>=8", + "opentelemetry-sdk>=1.25", + "agentrust-trace-tests>=0.6.2", +] + +[tool.setuptools] +packages = ["anthropic_tool_loop_trace"] diff --git a/integrations/anthropic-tool-loop/tests/test_anthropic_interop.py b/integrations/anthropic-tool-loop/tests/test_anthropic_interop.py new file mode 100644 index 0000000..eed661d --- /dev/null +++ b/integrations/anthropic-tool-loop/tests/test_anthropic_interop.py @@ -0,0 +1,91 @@ +"""The example's tool loop driven by the released `anthropic` SDK, offline. + +The SDK's own client, request building and response parsing run for real; only the HTTP transport +is replaced (MockTransport from the SDK's own HTTP library), so no API key or network is needed. A renamed field on Message, +Usage or ToolUseBlock fails here rather than silently producing empty spool rows. +""" + +from __future__ import annotations + +import importlib.util +import json +import pathlib +import sys + +import pytest + +anthropic = pytest.importorskip("anthropic") +from anthropic import _base_client # noqa: E402 + +# Use whichever HTTP library this SDK release is built on (httpx in older releases, httpx2 in newer). +httpx = getattr(_base_client, "httpx2", None) or _base_client.httpx +pytest.importorskip("agentrust_telemetry") + +from agentrust_trace import generate_key # noqa: E402 + +HERE = pathlib.Path(__file__).resolve().parents[1] +sys.path.insert(0, str(HERE)) +spec = importlib.util.spec_from_file_location("offline_tool_loop", HERE / "examples" / "offline_tool_loop.py") +example = importlib.util.module_from_spec(spec) +spec.loader.exec_module(example) + +from anthropic_tool_loop_trace.recorder import Recorder # noqa: E402 + + +def _message(stop_reason, content, cache_read=0): + return {"id": "msg_01", "type": "message", "role": "assistant", "model": example.MODEL, + "content": content, "stop_reason": stop_reason, "stop_sequence": None, + "usage": {"input_tokens": 400, "output_tokens": 30, "cache_read_input_tokens": cache_read, + "cache_creation_input_tokens": 0}} + + +def _tool_use(tool_id, name, args): + return {"type": "tool_use", "id": tool_id, "name": name, "input": args} + + +SCRIPT = [ + _message("tool_use", [_tool_use("toolu_01", "inventory_lookup", {"sku": "SKU-118"}), + _tool_use("toolu_02", "purchase_orders", + {"action": "create", "supplier": example.SUPPLIER, "qty": 20})]), + _message("end_turn", [{"type": "text", "text": "A buyer has to create it."}], cache_read=380), + _message("tool_use", [_tool_use("toolu_03", "purchase_orders", + {"action": "create", "supplier": example.SUPPLIER, "qty": 20})]), + _message("end_turn", [{"type": "text", "text": "Created."}], cache_read=380), +] + + +def test_real_sdk_loop_records_and_signs(tmp_path): + requests = [] + script = list(SCRIPT) + + def handler(request): + requests.append(json.loads(request.content)) + return httpx.Response(200, json=script.pop(0)) + + client = anthropic.Anthropic(api_key="offline-test-key", max_retries=0, + http_client=httpx.Client(transport=httpx.MockTransport(handler))) + spool = str(tmp_path / "spool.jsonl") + assert example.answer(client, "requester", "Stock? Order 20.", spool) == "A buyer has to create it." + assert example.answer(client, "buyer", "Go ahead.", spool) == "Created." + assert not script + + # The denied call went back to the model as an error result; the allowed one as data. + results = {b["tool_use_id"]: b for m in requests[1]["messages"] if m["role"] == "user" + and isinstance(m["content"], list) for b in m["content"]} + assert results["toolu_02"]["is_error"] is True and "is_error" not in results["toolu_01"] + + rows = [json.loads(line) for line in pathlib.Path(spool).read_text().splitlines()] + model_rows = [r for r in rows if r["kind"] == "model.call"] + assert [r["stop_reason"] for r in model_rows] == ["tool_use", "end_turn", "tool_use", "end_turn"] + assert model_rows[1]["cache_read"] == 380 and model_rows[0]["model"] == example.MODEL + assert [r["tool_use_id"] for r in rows if r["kind"] == "tool.call"] == ["toolu_01", "toolu_02", "toolu_03"] + assert "Northwind" not in pathlib.Path(spool).read_text() + + recorder = Recorder(spool=spool, data_dir=tmp_path / "data", signing_key=generate_key(), + catalog=example.CATALOG, enforcement_mode="enforce", + policy_bundle_digest="sha256:" + "ab" * 32, build_digest="sha256:" + "cd" * 32, + prompt_classification="internal") + first, second = (recorder.run(run_id)["run"] for run_id in recorder.poll()) + assert first["verdict"] == "contraindicated" and first["denies"] == 1 + assert second["verdict"] == "affirming" and second["writes"] == 1 + assert first["signed"] and second["signed"] diff --git a/integrations/anthropic-tool-loop/tests/test_anthropic_recorder.py b/integrations/anthropic-tool-loop/tests/test_anthropic_recorder.py new file mode 100644 index 0000000..f5a200d --- /dev/null +++ b/integrations/anthropic-tool-loop/tests/test_anthropic_recorder.py @@ -0,0 +1,307 @@ +"""The recorder: spool rows become AgenTrust events, OpenTelemetry spans and one signed TRACE record +per finished run. A denied call is visible in the appraisal, an unfinished run is never signed, and +an edited record fails verification.""" + +from __future__ import annotations + +import copy +import json +import pathlib +import sys +import time +from types import SimpleNamespace + +import pytest + +pytest.importorskip("agentrust_telemetry") +pytest.importorskip("agentrust_trace") + +from agentrust_trace import generate_key # noqa: E402 +from opentelemetry.sdk.trace import TracerProvider # noqa: E402 +from opentelemetry.sdk.trace.export import SimpleSpanProcessor # noqa: E402 +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter # noqa: E402 + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from anthropic_tool_loop_trace import Run, ToolCatalog, ToolInfo # noqa: E402 +from anthropic_tool_loop_trace.recorder import Recorder, source_digest, verify # noqa: E402 + +SECRET = "Acme Wholesale PO 4471 card 4111-1111-1111-1111" +MODEL = "claude-sonnet-5-5" +AGENT = "spiffe://example.org/agent/purchasing" +CATALOG = ToolCatalog( + {"inventory_lookup": ToolInfo("inventory-db", "database", "internal"), + "purchase_orders": ToolInfo("erp", "erp", "confidential", write=lambda verb: verb not in {"get", "list"}), + "crm_*": ToolInfo("crm", "crm", "confidential")}, + taxonomy="example.v1", order=("public", "internal", "confidential")) + + +def _response(stop, i=100, o=50, r=1000, w=200): + return SimpleNamespace(model=MODEL, stop_reason=stop, usage=SimpleNamespace( + input_tokens=i, output_tokens=o, cache_read_input_tokens=r, cache_creation_input_tokens=w)) + + +def _block(name, args, tool_id): + return SimpleNamespace(type="tool_use", id=tool_id, name=name, input=args) + + +@pytest.fixture +def env(tmp_path): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + spool = tmp_path / "spool.jsonl" + + def make(**overrides): + kwargs = dict(spool=spool, data_dir=tmp_path / "data", signing_key=key, catalog=CATALOG, + enforcement_mode="enforce", policy_bundle_digest="sha256:" + "ab" * 32, + build_digest="sha256:" + "cd" * 32, prompt_classification="internal", + tracer_provider=provider, + prices={MODEL: (3.0, 15.0, 0.30, 3.75)}, pricing_version="example-2026-10") + kwargs.update(overrides) + return Recorder(**kwargs) + + key = generate_key() + return SimpleNamespace(rec=make(), make=make, exporter=exporter, spool=str(spool), tmp=tmp_path) + + +def _run(env, **kw): + return Run(AGENT, "purchasing", MODEL, toolset=["inventory_lookup", "purchase_orders"], + action_field="action", spool=env.spool, **kw) + + +def _clean_run(env): + run = _run(env, workflow_id="ticket-12") + t = time.time_ns + run.model_call(0, t(), _response("tool_use")) + run.tool_call(0, _block("inventory_lookup", {"sku": SECRET}, "toolu_1"), t(), policy="purchasing_gate", + result={"rows": [SECRET]}, gate_ns=1200) + run.model_call(1, t(), _response("end_turn")) + run.end("end_turn") + return run + + +def _events(rec, event_type): + return [e for e in rec["events"] if e["event_type"] == event_type] + + +def test_clean_run_is_signed_affirming_and_verifies(env): + run = _clean_run(env) + assert env.rec.poll() == [run.id] + rec = env.rec.run(run.id) + s = rec["run"] + assert s["verdict"] == "affirming" and s["signed"] and s["complete"] + assert s["rounds"] == 2 and s["tool_calls"] == 1 and s["systems"] == ["inventory-db"] + assert s["data_class"] == "internal" and s["sent_to_model"] == "internal" + assert s["cost_usd"] == pytest.approx(2 * (100 * 3 + 50 * 15 + 1000 * 0.30 + 200 * 3.75) / 1e6) + assert env.rec.verify(rec["trace"])["ok"] is True + assert verify(rec["trace"], env.rec.public_jwk)["ok"] is True + trace = rec["trace"] + assert trace["origin"]["kind"] == "log-import" and trace["subject"] == AGENT + assert trace["model"] == {"provider": "anthropic", "model_id": MODEL} + assert trace["policy"] == {"bundle_hash": "sha256:" + "ab" * 32, "enforcement_mode": "enforce"} + assert trace["runtime"]["platform"] == "software-only" + + types = [e["event_type"] for e in rec["events"]] + assert types.count("usage.recorded") == 3 # two rounds + the agent_run rollup + assert types.count("policy.decision") == 2 # toolset + the tool call's gate + assert {"action.executed", "data_flow.observed"} <= set(types) + assert all(e.get("workflow_id") == "ticket-12" for e in rec["events"]) + toolset = [e for e in _events(rec, "policy.decision") if e["policy"]["policy_id"] == "toolset"][0] + assert toolset["decision"] == "allow" and toolset["action_type"] == "agent.invoke" + gate = [e for e in _events(rec, "policy.decision") if e["policy"]["policy_id"] == "purchasing_gate"][0] + assert gate["evaluation_duration_ns"] == 1200 + action = _events(rec, "action.executed")[0] + assert action["policy_event_id"] == gate["event_id"] and action["operation"] == "read" + + sent = [e for e in _events(rec, "data_flow.observed") if e["direction"] == "send"][0] + assert sent["destination"] == {"kind": "model-provider", "id": "anthropic"} + assert sent["policy_decision"] == "not_evaluated" + + spans = env.exporter.get_finished_spans() + names = {sp.name for sp in spans} + assert {"invoke_agent purchasing", f"chat {MODEL}", "execute_tool inventory_lookup"} <= names + assert len({sp.context.trace_id for sp in spans}) == 1 + assert s["otel_trace_id"] == f"{spans[0].context.trace_id:032x}" + + +def test_denied_call_is_contraindicated_with_reason(env): + run = _run(env) + t = time.time_ns + run.model_call(0, t(), _response("tool_use")) + run.tool_call(0, _block("purchase_orders", {"action": "create", "supplier": SECRET}, "toolu_9"), t(), + decision="deny", policy="purchasing_gate", reason="WRITERS_ONLY", gate_ns=900) + run.model_call(1, t(), _response("end_turn")) + run.end("end_turn") + env.rec.poll() + rec = env.rec.run(run.id) + s = rec["run"] + assert s["signed"] and s["verdict"] == "contraindicated" + assert rec["trace"]["appraisal"]["status"] == "contraindicated" + assert s["denies"] == 1 and s["writes"] == 0 + assert s["tools"][0]["write"] is True and s["tools"][0]["reason"] == "WRITERS_ONLY" + deny = [e for e in _events(rec, "policy.decision") if e["decision"] == "deny"][0] + assert deny["reason_codes"] == ["WRITERS_ONLY"] and deny["action_type"] == "erp.write" + assert _events(rec, "action.executed")[0]["outcome"] == "denied" + assert not [e for e in _events(rec, "data_flow.observed") if e["direction"] == "write"] + + +def test_allowed_write_is_a_write_flow(env): + run = _run(env) + t = time.time_ns + run.model_call(0, t(), _response("tool_use")) + run.tool_call(0, _block("purchase_orders", {"action": "create"}, "toolu_3"), t(), result={"po": 1}) + run.model_call(1, t(), _response("end_turn")) + run.end("end_turn") + env.rec.poll() + rec = env.rec.run(run.id) + flow = [e for e in _events(rec, "data_flow.observed") if e["direction"] == "write"][0] + assert flow["destination"] == {"kind": "erp", "id": "erp"} and flow["classification"]["value"] == "confidential" + assert rec["run"]["writes"] == 1 and rec["trace"]["data_class"] == "confidential" + assert _events(rec, "action.executed")[0]["operation"] == "create" + + +def test_chat_only_run_is_signed(env): + run = _run(env) + run.model_call(0, time.time_ns(), _response("end_turn")) + run.end("end_turn") + env.rec.poll() + s = env.rec.run(run.id)["run"] + assert s["signed"] and s["verdict"] == "affirming" and s["tool_calls"] == 0, s["trace_error"] + + +def test_unfinished_run_is_sealed_incomplete_and_unsigned(env): + run = _run(env) + run.model_call(0, time.time_ns(), _response("tool_use")) + assert env.rec.poll() == [] # still running + assert env.rec.poll(final=True) == [run.id] + rec = env.rec.run(run.id) + assert rec["trace"] is None + s = rec["run"] + assert s["signed"] is False and s["complete"] is False and s["verdict"] == "unsigned" + assert "never finished" in s["trace_error"] + + +def test_idle_run_is_sealed_unsigned(env): + rec = env.make(idle_seconds=0) + run = _run(env) + run.model_call(0, 1, _response("tool_use")) # started long ago + assert rec.poll() == [run.id] + assert rec.run(run.id)["run"]["signed"] is False + + +def test_tampered_record_fails_verification(env): + run = _clean_run(env) + env.rec.poll() + original = env.rec.run(run.id)["trace"] + for path, value in ((("appraisal", "status"), "warning"), (("data_class",), "public"), + (("policy", "enforcement_mode"), "advisory"), (("subject",), "spiffe://example.org/x")): + record = copy.deepcopy(original) + target = record + for key in path[:-1]: + target = target[key] + target[path[-1]] = value + assert env.rec.verify(record)["ok"] is False, path + other = generate_key() + assert env.make(signing_key=other).verify(original)["ok"] is False # someone else's key + + +def test_metadata_only_end_to_end(env): + run = _clean_run(env) + env.rec.poll() + everything = [pathlib.Path(env.spool)] + [p for p in (env.tmp / "data").rglob("*") if p.is_file()] + for path in everything: + assert "Acme" not in path.read_text(), path + for sp in env.exporter.get_finished_spans(): + assert "Acme" not in json.dumps(dict(sp.attributes), default=str) + assert env.rec.run(run.id)["run"]["signed"] + + +def test_state_survives_a_restart_mid_run(env): + run = _run(env) + run.model_call(0, time.time_ns(), _response("end_turn")) + env.rec.poll() + again = env.make() + assert run.id in again.pending + run.end("end_turn") + assert again.poll() == [run.id] + assert again.run(run.id)["run"]["signed"] is True + + +def test_half_written_line_waits(env): + run = _clean_run(env) + with open(env.spool, "a") as fh: + fh.write('{"run_id":"run-' + "a" * 32 + '","kind":"run.st') + assert env.rec.poll() == [run.id] + assert env.rec.pending == {} + with open(env.spool, "a") as fh: + fh.write('art","t":%d,"agent_id":"x","agent_name":"x","model":"m","toolset":[]}\n' % time.time_ns()) + env.rec.poll() + assert "run-" + "a" * 32 in env.rec.pending + + +def test_late_rows_do_not_reopen_a_sealed_record(env): + run = _clean_run(env) + env.rec.poll() + signed = env.rec.run(run.id)["trace"] + run.ended = False + run.end("end_turn") # a duplicate end arrives later + assert env.rec.poll() == [] + assert env.rec.run(run.id)["trace"] == signed + + +def test_without_opentelemetry_provider_events_still_record(env): + rec = env.make(tracer_provider=None) + run = _clean_run(env) + rec.poll() + stored = rec.run(run.id) + assert stored["run"]["signed"] and stored["run"]["otel_trace_id"] is None + assert all("trace_id" not in e for e in stored["events"]) + + +def test_configuration_is_explicit(env, tmp_path): + with pytest.raises(ValueError): + env.make(enforcement_mode="sometimes") + with pytest.raises(ValueError): + env.make(policy_bundle_digest="abc") + with pytest.raises(ValueError): + env.make(prompt_classification="secret") + with pytest.raises(TypeError): + Recorder(spool=env.spool, data_dir=tmp_path, signing_key=generate_key(), catalog=CATALOG, + policy_bundle_digest="sha256:" + "ab" * 32, build_digest="sha256:" + "cd" * 32, + prompt_classification="internal") # no enforcement_mode + + +def test_source_digest_changes_with_code(tmp_path): + (tmp_path / "gate.py").write_text("ALLOW = True\n") + first = source_digest(tmp_path) + (tmp_path / "gate.py").write_text("ALLOW = False\n") + assert source_digest(tmp_path) != first and first.startswith("sha256:") + + +def test_signed_record_passes_trace_conformance_level_0(env, tmp_path): + runner = pytest.importorskip("click.testing") + cli = pytest.importorskip("trace_tests.cli") + run = _clean_run(env) + env.rec.poll() + path = tmp_path / "record.json" + path.write_text(json.dumps(env.rec.run(run.id)["trace"])) + result = runner.CliRunner().invoke(cli.main, ["verify", "--record", str(path), "--level", "0"]) + assert result.exit_code == 0, result.output + assert "Ed25519 signature verified" in result.output + + +def test_a_malformed_run_is_logged_and_does_not_stop_the_others(env, caplog): + bad = "run-" + "b" * 32 + with open(env.spool, "a") as fh: + fh.write(json.dumps({"run_id": bad, "kind": "run.start", "t": time.time_ns(), "agent_id": AGENT}) + "\n") + fh.write(json.dumps({"run_id": bad, "kind": "tool.call", "start": time.time_ns(), "end": time.time_ns(), + "tool": "purchase_orders"}) + "\n") # no decision, digest or outcome + fh.write(json.dumps({"run_id": bad, "kind": "run.end", "t": time.time_ns(), "stop": "end_turn"}) + "\n") + fh.write("not json\n") + good = _clean_run(env) + with caplog.at_level("ERROR", logger="anthropic_tool_loop_trace.recorder"): + done = env.rec.poll() + assert set(done) == {bad, good.id} + assert env.rec.run(good.id)["run"]["signed"] and env.rec.run(bad) is None + assert any(bad in r.getMessage() for r in caplog.records) diff --git a/integrations/anthropic-tool-loop/tests/test_anthropic_spool.py b/integrations/anthropic-tool-loop/tests/test_anthropic_spool.py new file mode 100644 index 0000000..e3a6c4b --- /dev/null +++ b/integrations/anthropic-tool-loop/tests/test_anthropic_spool.py @@ -0,0 +1,158 @@ +"""The in-process writer: standard library only, metadata only, never raises into the agent. + +Runs without agentrust-telemetry or agentrust-trace installed, and on Python 3.9. +""" + +from __future__ import annotations + +import json +import logging +import pathlib +import sys +from types import SimpleNamespace + +import pytest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from anthropic_tool_loop_trace import Run, ToolCatalog, ToolInfo, digest # noqa: E402 +from anthropic_tool_loop_trace import spool as spool_module # noqa: E402 + +SECRET = "Acme Wholesale card 4111-1111-1111-1111" + + +def _response(stop="tool_use"): + return SimpleNamespace(model="claude-sonnet-5-5", stop_reason=stop, usage=SimpleNamespace( + input_tokens=100, output_tokens=20, cache_read_input_tokens=5, cache_creation_input_tokens=7)) + + +def _block(tool="crm_search", args=None, tool_id="toolu_1"): + return SimpleNamespace(type="tool_use", id=tool_id, name=tool, input=args or {"q": SECRET}) + + +def _rows(path): + return [json.loads(line) for line in pathlib.Path(path).read_text().splitlines()] + + +def test_rows_carry_metadata_and_no_content(tmp_path): + path = tmp_path / "spool.jsonl" + run = Run("spiffe://example.org/agent/a", "a", "claude-sonnet-5-5", toolset=["crm_search", "crm_update"], + workflow_id="ticket-7", action_field="action", spool=str(path)) + run.model_call(0, 1, _response()) + run.tool_call(0, _block(args={"action": "update", "note": SECRET}), 2, result={"rows": [SECRET]}, gate_ns=5) + run.tool_call(0, _block(tool="crm_update", tool_id="toolu_2", args={"action": SECRET}), 3, + decision="deny", policy="writers_only", reason="WRITERS_ONLY") + run.end("end_turn") + + text = path.read_text() + assert SECRET not in text and "Acme" not in text + start, model, allowed, denied, end = _rows(path) + assert start["kind"] == "run.start" and start["toolset"] == ["crm_search", "crm_update"] + assert start["workflow_id"] == "ticket-7" and start["agent_name"] == "a" + assert model["input"] == 100 and model["cache_read"] == 5 and model["cache_write"] == 7 + assert allowed["action"] == "update" and allowed["outcome"] == "success" and allowed["result_chars"] > 0 + assert allowed["input_digest"] == digest({"action": "update", "note": SECRET}) + # A free-text value in the action field is not a verb and is not recorded. + assert denied["action"] is None and denied["outcome"] == "denied" and denied["reason"] == "WRITERS_ONLY" + assert end["stop"] == "end_turn" + assert len({r["run_id"] for r in _rows(path)}) == 1 + + +def test_action_field_is_opt_in(tmp_path): + path = tmp_path / "spool.jsonl" + run = Run("agent", "a", "m", spool=str(path)) + run.tool_call(0, _block(args={"action": "create"}), 1) + assert _rows(path)[-1]["action"] is None + + +def test_reason_and_policy_are_codes_not_messages(tmp_path): + path = tmp_path / "spool.jsonl" + run = Run("agent", "a", "m", spool=str(path)) + run.tool_call(0, _block(), 1, decision="deny", policy="gate", reason="Only Dana may send POs to " + SECRET) + row = _rows(path)[-1] + assert row["reason"] == "UNSPECIFIED" and SECRET not in json.dumps(row) + + +def test_outcomes(tmp_path): + path = tmp_path / "spool.jsonl" + run = Run("agent", "a", "m", spool=str(path)) + run.tool_call(0, _block(), 1, error_type="TimeoutError") + run.tool_call(0, _block(), 1, result={"error": "not found"}) + run.tool_call(0, _block(), 1, decision="maybe") # anything but allow is treated as deny + errored, tool_error, odd = _rows(path)[1:] + assert errored["outcome"] == "error" and errored["error_type"] == "TimeoutError" + assert tool_error["outcome"] == "success" and tool_error["tool_error"] is True + assert odd["decision"] == "deny" and odd["outcome"] == "denied" + + +def test_end_is_written_once(tmp_path): + path = tmp_path / "spool.jsonl" + run = Run("agent", "a", "m", spool=str(path)) + run.end("end_turn") + run.end("max_rounds") + assert [r["kind"] for r in _rows(path)] == ["run.start", "run.end"] + + +def test_dict_shaped_responses_work(tmp_path): + path = tmp_path / "spool.jsonl" + run = Run("agent", "a", "m", spool=str(path)) + run.model_call(0, 1, {"model": "m", "stop_reason": "end_turn", "usage": {"input_tokens": 3, "output_tokens": 4}}) + run.tool_call(0, {"name": "t", "id": "toolu_x", "input": {}}, 1) + model, tool = _rows(path)[1:] + assert model["input"] == 3 and model["output"] == 4 and tool["tool"] == "t" + + +def test_writer_fails_open(tmp_path, caplog): + blocked = tmp_path / "is-a-directory" + blocked.mkdir() + with caplog.at_level(logging.WARNING, logger="anthropic_tool_loop_trace"): + run = Run("agent", "a", "m", toolset=[object()], spool=str(blocked)) + run.model_call(0, 1, object()) # no usage, no model + run.model_call("not-a-number", 1, _response()) # bad round + run.tool_call(0, None, 1) # no block at all + run.end("end_turn") + assert run.ended + assert any("write failed" in r.message for r in caplog.records) + + +def test_writer_survives_a_broken_lock(tmp_path, monkeypatch): + class Broken: + def __enter__(self): + raise RuntimeError("lock broke") + + def __exit__(self, *exc): + return False + + monkeypatch.setattr(spool_module, "_LOCK", Broken()) + run = Run("agent", "a", "m", spool=str(tmp_path / "s.jsonl")) + run.tool_call(0, _block(), 1) + run.end("end_turn") + assert not (tmp_path / "s.jsonl").exists() + + +def test_default_path_from_environment(monkeypatch, tmp_path): + monkeypatch.setenv(spool_module.SPOOL_ENV, str(tmp_path / "env.jsonl")) + Run("agent", "a", "m").end("end_turn") + assert len(_rows(tmp_path / "env.jsonl")) == 2 + + +def test_digest_is_canonical(): + assert digest({"b": 1, "a": [1, 2]}) == digest({"a": [1, 2], "b": 1}) + assert digest({"a": 1}) != digest({"a": 2}) + + +def test_catalog_lookup_and_conservative_default(): + catalog = ToolCatalog( + {"crm_*": ToolInfo("crm", "crm", "confidential"), "crm_public_*": ToolInfo("site", "saas", "public"), + "orders": ToolInfo("erp", "erp", "internal", write=lambda verb: verb not in {"get", None})}, + taxonomy="t.v1", order=("public", "internal", "confidential")) + assert catalog.lookup("crm_search").system == "crm" + assert catalog.lookup("crm_public_faq").system == "site" # longest prefix wins + assert catalog.is_write("orders", "create") and not catalog.is_write("orders", "get") + assert not catalog.is_write("crm_search") + unknown = catalog.lookup("shell_exec") + assert unknown.system == "uncatalogued" and unknown.classification == "confidential" + assert catalog.is_write("shell_exec") + assert catalog.highest(["public", "internal"]) == "internal" + with pytest.raises(ValueError): + ToolCatalog({"x": ToolInfo("x", classification="secret")}, taxonomy="t", order=("public",)) diff --git a/marketplace/catalog.json b/marketplace/catalog.json index 9874477..e953885 100644 --- a/marketplace/catalog.json +++ b/marketplace/catalog.json @@ -1,6 +1,6 @@ { "catalog_version": 1, - "count": 47, + "count": 48, "integrations": [ { "name": "Claude Code", @@ -273,6 +273,30 @@ "case-law" ] }, + { + "name": "Anthropic SDK tool loop", + "package_name": "Anthropic SDK tool loop", + "vendor": "Disruptive Beverages", + "description": "Records a Claude tool loop built on the anthropic Python SDK as AgenTrust Telemetry events, metadata only, and signs one log-import TRACE record per run.", + "path": "integrations/anthropic-tool-loop", + "url": "https://github.com/agentrust-io/integrations/tree/main/integrations/anthropic-tool-loop", + "homepage": null, + "repository": "https://github.com/agentrust-io/integrations", + "tier": "community", + "stack": [ + "TRACE" + ], + "category": "Frameworks", + "mark": "AN", + "featured": null, + "keywords": [ + "anthropic", + "claude", + "tool-use", + "telemetry", + "log-import" + ] + }, { "name": "Azure Secure Key Release", "package_name": "WCM Azure Secure Key Release",