wytness

Wytness wrapper for Microsoft's Agent Governance Toolkit (AGT). Adds the two security properties AGT does not ship:

• Non-repudiation — per-event Ed25519 signature with a customer-held private key. Asymmetric verification: anyone with the public key can verify, no shared secret.
• Zero-knowledge PII — HMAC-SHA256 pseudonyms + X25519 + ChaCha20-Poly1305 token-map encryption. Wytness never sees raw PII.

These compose with AGT's own primitives (Ed25519 agent identity, SHA-256 hash chain). See docs/agt-wrapper-design.md §2.5 for the full layered model.

Mirrors wytness-ai (Python) — same wire format, same init() / shutdown() lifecycle.

Status

Current version: 1.8.0. Depends on @microsoft/[email protected] (hard dependency, installed automatically). Mirrors wytness-ai 1.8.0 (Python) — same wire format, same env-var contract.

init({ autoWire: true }) (default) monkey-patches AuditLogger.prototype.log so every audit entry AGT chains is forwarded into the Wytness envelope path with full 3-layer evidence (AGT identity + AGT chain + Wytness envelope, envelope_version: 2 — signature covers canonical-JSON of {data, id, time, type}). The TS wrapper transforms AGT-TS's camelCase fields into the canonical snake_case shape per docs/agt-wrapper-design.md §13; the Wytness data payload is byte-comparable to the Python wrapper for the same canonical inputs.

Public surface: init(), shutdown(), WytnessConfig, WytnessAuditSink, WytnessEventSink, SessionTokenizer, stats(), health(), wireStatus(), AuditLogger (re-exported from AGT), and the typed exception hierarchy (WytnessSDKError → WytnessConfigError / WytnessInitError / WytnessTransportError).

Tests: a vitest suite covering config validation (XOR PII pairing, env-var fallback), envelope signing + tamper-detection, PII pseudonymisation + auto-attach, transport batching + retry + dead-letter file + jittered backoff, init/shutdown idempotency, AGT-TS wrap against 5.0.0, wire-format byte parity with the Python SDK fixture, and stats/health pre-init safety.

Install

npm install @wytness/ai

@microsoft/[email protected] is a hard dependency — installed automatically. Targets Node 18+ (uses the global fetch).

Hello world

import { init, shutdown, setNextEntryExtras, AuditLogger } from "@wytness/ai";

await init({
  apiKey: process.env.WYTNESS_API_KEY,
  piiPubkey: process.env.WYTNESS_PII_PUBKEY,     // browser-generated X25519
  piiSecret: process.env.WYTNESS_PII_SECRET,     // browser-generated HMAC
  signingKey: process.env.WYTNESS_SIGNING_KEY,   // browser-generated Ed25519
  piiFields: ["customer.email"],
});

// AGT-TS owns identity + chain. The wrapper has already monkey-patched
// AuditLogger.prototype.log — every entry below also flows to Wytness.
const logger = new AuditLogger();

// Stash the Wytness extras AGT-TS doesn't accept directly (data, outcome, …).
setNextEntryExtras({
  data: { customer: { email: "[email protected]" }, query: "best espresso" },
  outcome: "success",
  resource: "https://example.com",
});

logger.log({
  agentId: "audit-test-agent",
  action: "tool_invocation:search_web",   // colon-split → event_type + action
  decision: "allow",                       // → policy_decision
});

await shutdown();

PII is pseudonymised before egress, the canonical-JSON envelope is signed with your Ed25519 key, and the batch lands at api.wytness.ai/ingest. Failed POSTs land in ~/.wytness/wytness-deadletter.jsonl — recover them with wytness-ai replay-deadletter (add --dry-run to list first): it re-POSTs the already-signed envelopes through the normal transport and is safe to re-run, because the backend deduplicates replayed event ids.

AuditLogger is re-exported from @wytness/ai, so one import line covers both lifecycle and audit. The original import { AuditLogger } from "@microsoft/agent-governance-sdk" is identical and still works.

Browser-generated keys come from the Wytness dashboard's Keys page (Signing Key, PII Encryption Key, and PII HMAC Key cards); the dashboard never sees your private halves.

Automatic LLM capture

Wrap your model client and every Anthropic Messages API (including the beta namespace) and OpenAI Chat Completions and Responses call — streaming included — is recorded automatically — prompt, reply, reasoning (Anthropic extended thinking), latency, and token usage — with no AuditLogger().log(...) in your code. Token usage needs no opt-in: on OpenAI streams the SDK requests it and hides the resulting usage-only chunk, so your loop is unchanged. Other providers are not captured. Captured calls ride the same signed, secret-scrubbed sink as manual audit entries; when PII pseudonymisation is enabled the conversation is tokenised in-process before it leaves the machine.

import Anthropic from "@anthropic-ai/sdk";
import { init, shutdown, wrapAnthropic } from "@wytness/ai";

await init({ agentDid: "did:web:acme.example:support-bot" });

const client = wrapAnthropic(new Anthropic());   // capture every call on this client
await client.messages.create({                    // recorded automatically
  model: "claude-haiku-4-5",
  max_tokens: 512,
  messages: [{ role: "user", content: "Summarise ticket #4821" }],
});

await shutdown();

OpenAI is identical — wrap the client after init():

import OpenAI from "openai";
import { wrapOpenAI } from "@wytness/ai";

const openai = wrapOpenAI(new OpenAI());
await openai.chat.completions.create({
  model: "gpt-4o-mini",
  messages: [{ role: "user", content: "Summarise ticket #4821" }],
});

Pass wrapAnthropic(client, { agentDid }) to attribute a client's captured calls to a specific agent; it defaults to init()'s agentDid. Anthropic and OpenAI are supported today. Streaming calls are captured too: create({stream: true}) resolves to a transparent tee that records the assembled response when the stream finishes (or errors / breaks early), and messages.stream(...) captures from its own finalMessage event — the MessageStream object and its helpers are untouched. Streamed captures carry parameters.stream: true. OpenAI streamed usage needs no opt-in — the SDK sets stream_options when you have not, and never overrides one you set yourself.

Parity note. The Python SDK auto-patches the provider at init(wrap=…); TypeScript uses an explicit wrapAnthropic(client) / wrapOpenAI(client) because the provider SDK classes aren't safely patchable at init. The recorded event is identical — only the one-line setup differs.

Capture-status reporting. The SDK periodically reports which surfaces it is recording to Wytness so your dashboard can state what is watched. The report carries surface names, capture states, and delivery counters — never prompts, payloads, or credentials — over the same endpoint and API key as your events. Fail-soft (a failed report is retried later, capture is never affected); the timer is unref()'d and dryRun disables it.

CLI

npx @wytness/ai <command> (bin name wytness-ai). All commands read the same WYTNESS_* env vars and never generate keys.

| Command | What it does | |---|---| | wytness-ai verify | Round-trip one signed event end-to-end and confirm it arrives in the dashboard. | | wytness-ai tail | Stream recent dashboard events to your terminal. Flags: --agent-id, --limit, --interval, --once, --json. | | wytness-ai validate-chain | Pull recent events and walk the AGT hash chain locally, proving link integrity without trusting the server. Flags: --agent-id, --limit, --json. |

Dry run

Set WYTNESS_DRY_RUN=1 (or init({ dryRun: true })) to build and Ed25519-sign every envelope but skip the POST to /ingest — the signed envelope is written to stderr instead. The init-time signing-key registration probe is skipped too, so it runs fully offline. Ideal for CI, unit tests, and local development.

Framework recipes

Where to call init() / shutdown() in Express, Next.js, Vercel, and AWS Lambda: docs/recipes.md.

Feature parity with wytness-ai (Python)

Both SDKs ship the same wire format (snake_case canonical shape, identical envelope structure, byte-comparable for the same logical input — verified by tests/integration/wireFormat.parity.test.ts). Behavioural surface differences:

| Surface | Python | TypeScript | Notes | |---|---|---|---| | AGT AuditLog auto-wire | ✓ — patches agentmesh.governance.audit.AuditLog | ✓ — patches AuditLogger.prototype.log from @microsoft/agent-governance-sdk | Both default auto_wire=True / autoWire: true. | | AGT GovernanceEvent auto-wire | ✓ — registers WytnessEventSink on GovernanceEventProcessor | ✗ — manual (getEventSink().emit(...)) | AGT-TS 5.0.0 has no GovernanceEventProcessor registry. Tracked as V2-AGT-TS-EVENT-SINK; will wire automatically when Microsoft ships the registry. | | Wire-format byte parity | ✓ | ✓ | Cross-language fixture checked in. | | Ed25519 envelope signing | ✓ | ✓ | Identical canonical-JSON encoding + sig scope. | | PII pseudonymisation | ✓ | ✓ | Identical HMAC + X25519 + ChaCha20-Poly1305 primitives. | | Framework adapter compatibility tests | ✓ — 12 integrations × 3 axes | ✗ — AGT-TS ships zero framework adapters in 5.0.0 | | | stats() + health() accessors | ✓ — wytness.stats() / health() | ✓ — wytnessAgt.stats() / health() | Same shape; snake_case (Py) vs camelCase (TS). |

Manual wiring (no monkey-patch)

If you'd rather wire explicitly, pass autoWire: false and call the audit sink directly. The sink accepts both AGT-TS camelCase and pre-canonical snake_case shapes — the canonical-shape transform runs automatically when it sees camelCase keys.

import { init, getAuditSink } from "@wytness/ai";
await init({ /* … */, autoWire: false });
const sink = getAuditSink()!;
sink.write({
  agentId: "agent-1",
  action: "tool_invocation:search_web",
  decision: "allow",
  hash: "deadbeef…",
  previousHash: "0".repeat(64),
  data: { query: "…" },
});

Wire format

Each POST batch is a JSON array of CloudEvents envelopes:

[
  {
    "specversion": "1.0",
    "id": "<uuid>",
    "source": "wytness",
    "type": "AuditEntry",
    "datacontenttype": "application/json",
    "time": "2026-05-22T00:00:00.000Z",
    "data": {
      "timestamp": "2026-05-22T00:00:00.000Z",
      "agent_id": "audit-test-agent",
      "event_type": "tool_invocation",
      "action": "search_web",
      "policy_decision": "allow",
      "entry_hash": "<AGT-supplied SHA-256 hex>",
      "previous_hash": "",
      "entry_id": "<uuid>",
      "data": { "query": "…", "customer": { "email": "EMAIL_<token>" } },
      "outcome": "success",
      "resource": "https://example.com"
    },
    "wytness_envelope": {
      "envelope_version": 2,
      "key_id": "<16 chars base64url SHA-256(pubkey)>",
      "signature_ed25519": "<base64 Ed25519 sig over canonicalJson({data, id, time, type})>",
      "pseudonymization_version": "1"
    }
  }
]

The data field is the canonical snake_case shape — byte-comparable to what wytness-ai (Python) emits for the same logical input. CH agt_did, agt_merkle_entry_hash, agt_merkle_previous_hash, policy_decision columns map 1:1.

Content-Type: application/vnd.wytness.agt+json — every envelope is signed (Wytness has no unsigned mode; signing is the product's value prop).

Idempotency / deduplication

The SDK gives you network-retry idempotency for free: when a POST fails and is retried, the same envelope id is re-sent, so the backend deduplicates the duplicate write.

The SDK does not automatically deduplicate customer-side retries — a logical action retried twice in your app code produces two distinct envelopes with two distinct entry_ids, which means two CH rows. If you need that guarantee, thread a stable per-action key through setNextEntryExtras:

setNextEntryExtras({
  entry_id: "tool-invocation-2026-06-01-uuid-abc",
  data: { customer: { email: "[email protected]" }, query: "best espresso" },
  outcome: "success",
});

logger.log({
  agentId: "my-first-agent",
  action: "tool_invocation:search",
  decision: "allow",
});

When entry_id is present at the boundary the SDK preserves it through canonicalisation; the backend then uses (org_id, entry_id) as the dedup tuple. Absent the field, the SDK synthesises a fresh UUID per write() call.

License

MIT. Built on @microsoft/agent-governance-sdk (MIT, Microsoft).