feat: add structured audit event system (#1027)

Add a complete audit event pipeline that emits structured, machine-
parseable events for every gateway request and IAM decision.

Schema and publisher:
- AuditEvent dataclass and notify-class queue (audit_events_queue)
- AuditPublisher utility: fire-and-forget emission with envelope
  (schema_version, event_id, event_type, timestamp, producer)
- New request_id and client_ip fields on IamRequest for correlation

Gateway (gateway.request events):
- aiohttp middleware assigns request_id, captures timing/status/sizes
  and emits an event after every HTTP request completes
- IamAuth.authenticate annotates the request with identity
- Main endpoint handlers annotate capability and workspace
- request_id and client_ip forwarded to IAM on authenticate/authorise

IAM service (iam.authenticate, iam.authorise, iam.management events):
- Emits iam.authenticate for resolve-api-key, login, anonymous auth
- Emits iam.authorise for authorise and authorise-many decisions
- Emits iam.management for user/workspace/key mutations
- All events include request_id for correlation with gateway events

Design: events land on a pub/sub notify topic — non-persistent,
per-subscriber delivery. If no audit consumer is deployed, events
are silently discarded. Storage, retention, and alerting are
consumer-side concerns outside this boundary.

Added unit tests for the publisher and gateway middleware, unit
tests for IAM audit emission, and a contract test for the AuditEvent
schema.

Tech spec: docs/tech-specs/audit-events.md

docs/tech-specs/audit-events.md

@@ -0,0 +1,402 @@
+---
+layout: default
+title: "Audit Events Technical Specification"
+parent: "Tech Specs"
+---
+
+# Audit Events Technical Specification
+
+## Overview
+
+This specification defines the audit event system for TrustGraph.
+Audit events provide a structured, complete record of security-
+relevant operations: API gateway invocations and IAM decisions.
+
+The design principle is: **emit everything, let consumers decide.**
+Audit events are cheap to produce (a pub/sub message per operation)
+and rich enough to support any downstream consumer — compliance
+dashboards, SIEM integration, anomaly detection, billing metering,
+or simple grep-based debugging.  This spec covers event production
+only.  Storage, retention, alerting, and presentation are
+deployment-specific concerns handled by consumers outside this
+boundary.
+
+## Motivation
+
+TrustGraph currently has operational logging (Python `logging` to
+stdout/Loki) but no structured audit trail.  Operational logs are
+unstructured, filtered by level, and designed for debugging — not
+for answering "who did what, when, and was it allowed?"
+
+Enterprise deployments need:
+
+- **Compliance evidence** — demonstrable record of access for
+  auditors.
+- **Incident investigation** — reconstruct what happened around a
+  security event.
+- **Anomaly detection** — feed structured events into monitoring
+  systems.
+- **Accountability** — attribute actions to identities across
+  workspaces.
+
+The current logging infrastructure cannot serve these needs because
+it is unstructured, inconsistently formatted, and interleaves
+debug noise with security-relevant signals.
+
+## Design Principles
+
+- **Complete.**  Every gateway request and every IAM decision emits
+  an event.  No sampling, no level-gating.  The pub/sub cost is
+  negligible; consumers filter what they need.
+
+- **Structured.**  Events are typed, versioned, machine-parseable
+  JSON objects with a fixed envelope and operation-specific payloads.
+  No free-text messages.
+
+- **Cheap to produce.**  Events land on a pub/sub topic.  No
+  synchronous writes, no blocking on consumer availability.  If no
+  consumer is subscribed, events are discarded by the broker — that
+  is acceptable.
+
+- **Rich.**  Events carry enough context to reconstruct the full
+  security narrative without correlating against operational logs.
+  Identity, workspace, capability, resource, outcome, timing,
+  client metadata.
+
+- **Immutable.**  Once emitted, an event is a fact.  Consumers may
+  filter, aggregate, or discard events, but never mutate them.
+
+- **Decoupled.**  Producers (gateway, IAM service) have no knowledge
+  of consumers.  The topic is fire-and-forget.  This keeps the
+  critical path fast and allows diverse consumer deployments.
+
+## Architecture
+
+### Event transport
+
+Audit events are published to a dedicated pub/sub topic, declared
+in the schema layer following the project's queue naming convention:
+
+```python
+audit_events_queue = queue('audit-events', cls='notify')
+```
+
+This produces the queue identifier `notify:tg:audit-events`, which
+each backend maps to its native topic format (e.g. Pulsar maps
+`notify` to `non-persistent://tg/notify/audit-events`).
+
+The `notify` class is the right fit: non-persistent, per-subscriber
+delivery, no competing-consumer semantics.  Audit event production
+must never block the gateway or IAM service.  Consumers that need
+durability persist events themselves on receipt.
+
+A single topic carries all event types, distinguished by the
+`event_type` field in the envelope.  This simplifies producer
+logic and allows consumers to subscribe once and filter client-side.
+
+### Producers
+
+Two components emit audit events:
+
+1. **API Gateway** — emits a `gateway.request` event for every
+   inbound HTTP/WebSocket request after the request completes
+   (or fails).
+
+2. **IAM Service** — emits `iam.authenticate` and `iam.authorise`
+   events for every authentication and authorisation decision.
+
+Both producers emit asynchronously — the event is published after
+the response is sent (gateway) or after the decision is returned
+(IAM).  Audit emission is never on the critical path.
+
+### Consumers
+
+Not defined by this spec.  Example consumers that deployments
+may wire up:
+
+- Append to an immutable log store (S3, Cassandra, ClickHouse).
+- Forward to a SIEM (Splunk, Elastic, Sentinel).
+- Aggregate for billing/metering.
+- Feed an anomaly detection model.
+- Write to stdout for development debugging.
+
+## Event Envelope
+
+Every audit event shares a common envelope:
+
+```json
+{
+  "schema_version": 1,
+  "event_id": "uuid-v4",
+  "event_type": "gateway.request",
+  "timestamp": "2026-07-05T14:23:01.123Z",
+  "producer": "api-gateway",
+  "payload": { ... }
+}
+```
+
+| Field | Type | Description |
+|---|---|---|
+| `schema_version` | int | Envelope schema version.  Consumers must ignore events with versions they don't understand. |
+| `event_id` | string | Globally unique event identifier (UUID v4). |
+| `event_type` | string | Dot-separated event type from the vocabulary below. |
+| `timestamp` | string | ISO 8601 UTC timestamp at event emission. |
+| `producer` | string | Component identity that emitted the event. |
+| `payload` | object | Event-type-specific structured data. |
+
+## Event Types
+
+### `gateway.request`
+
+Emitted by the API gateway for every completed request.
+
+```json
+{
+  "request_id": "uuid-v4",
+  "method": "POST",
+  "path": "/api/v1/flow/default/graph-rag",
+  "capability": "graph-rag:query",
+  "workspace": "production",
+  "identity": "user:mark",
+  "client_ip": "192.168.1.42",
+  "user_agent": "trustgraph-cli/2.6.11",
+  "status_code": 200,
+  "outcome": "success",
+  "duration_ms": 1423,
+  "request_size_bytes": 256,
+  "response_size_bytes": 4096,
+  "parameters": {
+    "collection": "default",
+    "entity_limit": 50
+  }
+}
+```
+
+| Field | Type | Description |
+|---|---|---|
+| `request_id` | string | Unique ID for this request, propagated to IAM events for correlation. |
+| `method` | string | HTTP method. |
+| `path` | string | Request path (no query string). |
+| `capability` | string | The capability required for this endpoint (from the capability vocabulary). |
+| `workspace` | string | Resolved workspace for this request. |
+| `identity` | string | Authenticated identity handle, or `"anonymous"` if unauthenticated. |
+| `client_ip` | string | Client IP address (may be from X-Forwarded-For). |
+| `user_agent` | string | Client User-Agent header. |
+| `status_code` | int | HTTP response status code. |
+| `outcome` | string | One of `success`, `denied`, `error`, `unauthenticated`. |
+| `duration_ms` | int | Request duration in milliseconds. |
+| `request_size_bytes` | int | Request body size. |
+| `response_size_bytes` | int | Response body size. |
+| `error` | string | Error category.  Present only when outcome is not `success`. |
+| `parameters` | object | Operation-specific parameters extracted from the request (not the full body — only semantically relevant fields). |
+
+### `iam.authenticate`
+
+Emitted by the IAM service for every authentication attempt.
+
+```json
+{
+  "request_id": "uuid-v4",
+  "credential_type": "api-key",
+  "identity": "user:mark",
+  "outcome": "success",
+  "client_ip": "192.168.1.42",
+  "key_id": "key-abc123"
+}
+```
+
+| Field | Type | Description |
+|---|---|---|
+| `request_id` | string | Correlates with the gateway request that triggered this authentication. |
+| `credential_type` | string | One of `api-key`, `jwt`, `login-password`. |
+| `identity` | string | Resolved identity on success, or `"unknown"` on failure. |
+| `outcome` | string | One of `success`, `failure`. |
+| `failure_reason` | string | Internal failure category (not exposed to clients): `invalid-key`, `expired-jwt`, `bad-signature`, `user-disabled`, `unknown-user`.  Present only on failure. |
+| `client_ip` | string | Forwarded from the gateway request. |
+| `key_id` | string | API key identifier (not the secret).  Present only on key-based auth. |
+
+**Note:** `failure_reason` is for the audit log only.  The client
+response is always the same masked error per the IAM contract's
+security rule.  The audit consumer sees the real reason; the
+attacker does not.
+
+### `iam.authorise`
+
+Emitted by the IAM service for every authorisation decision.
+
+```json
+{
+  "request_id": "uuid-v4",
+  "identity": "user:mark",
+  "capability": "graph-rag:query",
+  "workspace": "production",
+  "resource": "flow:default",
+  "outcome": "allow",
+  "evaluated_roles": ["workspace-user"],
+  "evaluation_time_us": 42
+}
+```
+
+| Field | Type | Description |
+|---|---|---|
+| `request_id` | string | Correlates with the gateway request. |
+| `identity` | string | Identity being authorised. |
+| `capability` | string | Capability being checked. |
+| `workspace` | string | Workspace scope of the resource. |
+| `resource` | string | Structured resource identifier. |
+| `outcome` | string | One of `allow`, `deny`. |
+| `denial_reason` | string | Why denied: `no-matching-role`, `capability-not-in-role`, `workspace-not-accessible`, `user-disabled`.  Present only on denial. |
+| `evaluated_roles` | list of string | Roles evaluated during the decision (OSS regime specific — other regimes may populate differently). |
+| `evaluation_time_us` | int | Time to evaluate the decision in microseconds. |
+
+### `iam.management`
+
+Emitted by the IAM service for administrative mutations.
+
+```json
+{
+  "request_id": "uuid-v4",
+  "actor": "user:admin",
+  "operation": "create-user",
+  "target_identity": "user:new-hire",
+  "target_workspace": "engineering",
+  "outcome": "success",
+  "details": {
+    "roles_assigned": ["workspace-user"]
+  }
+}
+```
+
+| Field | Type | Description |
+|---|---|---|
+| `request_id` | string | Correlates with the gateway request. |
+| `actor` | string | Identity performing the action. |
+| `operation` | string | IAM operation name (`create-user`, `delete-api-key`, `assign-role`, `create-workspace`, etc.). |
+| `target_identity` | string | Identity being acted upon.  Present only when applicable. |
+| `target_workspace` | string | Workspace being acted upon.  Present only when applicable. |
+| `outcome` | string | One of `success`, `error`. |
+| `details` | object | Operation-specific details (roles assigned, key created, etc.). |
+
+## Correlation
+
+All events from a single gateway request share the same
+`request_id`.  A typical request produces:
+
+1. One `gateway.request` event (after completion).
+2. One `iam.authenticate` event (credential validation).
+3. One or more `iam.authorise` events (capability checks).
+
+Consumers can reconstruct the full request lifecycle by grouping
+on `request_id`.
+
+## Implementation
+
+### Gateway changes
+
+The gateway emits `gateway.request` events.  Implementation:
+
+- Assign a UUID `request_id` at request entry.
+- Pass `request_id` and `client_ip` to the IAM service in the
+  `IamRequest` (new fields on the dataclass).
+- After the response is sent, publish the audit event to the
+  audit topic.  This is a non-blocking fire-and-forget publish.
+
+### IAM service changes
+
+The IAM service emits `iam.authenticate`, `iam.authorise`, and
+`iam.management` events.  Implementation:
+
+- Accept `request_id` and `client_ip` from the gateway on each
+  `IamRequest`.
+- After each decision or mutation, publish the corresponding audit
+  event.  Non-blocking.
+
+### Schema additions
+
+New queue declaration in `trustgraph-base/trustgraph/schema/`:
+
+```python
+from trustgraph.schema.core.topic import queue
+
+audit_events_queue = queue('audit-events', cls='notify')
+```
+
+New fields on `IamRequest`:
+
+```python
+@dataclass
+class IamRequest:
+    ...
+    request_id: str = ""
+    client_ip: str = ""
+```
+
+These are informational — the IAM service does not act on them
+beyond echoing them into audit events.
+
+### Pub/sub producer
+
+A lightweight audit publisher utility in `trustgraph-base`:
+
+```python
+class AuditPublisher:
+    def __init__(self, producer):
+        self.producer = producer
+
+    async def emit(self, event_type, payload):
+        event = {
+            "schema_version": 1,
+            "event_id": str(uuid4()),
+            "event_type": event_type,
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "producer": self.component_name,
+            "payload": payload,
+        }
+        await self.producer.send(json.dumps(event).encode())
+```
+
+The publisher is instantiated once per component and shared across
+request handlers.
+
+## What This Spec Does Not Cover
+
+- **Storage.**  Where audit events are persisted, for how long,
+  and in what format.  Deployment-specific.
+- **Alerting.**  What conditions trigger alerts.  Consumer logic.
+- **Retention policy.**  How long events are kept.  Compliance-
+  dependent.
+- **UI.**  Audit log viewers, dashboards, search interfaces.
+- **Filtering/routing.**  Topic partitioning, consumer-side
+  filtering, event routing to different backends.
+- **Redaction.**  PII handling in audit events (may be needed for
+  GDPR — a future concern for enterprise consumers).
+
+These are all consumer-side concerns.  The value of this boundary
+is that producers remain simple and fast while consumers can be
+as sophisticated as the deployment requires.
+
+## Open Questions
+
+- **Should WebSocket upgrade events emit separately from per-message
+  events?**  Current proposal: one `gateway.request` per WebSocket
+  session (on close), with `duration_ms` covering the full session.
+  Per-message audit for long-lived sockets (e.g. streaming RAG) may
+  be needed for metering but adds volume.
+
+- **Should `parameters` in `gateway.request` be standardised per
+  endpoint, or free-form?**  Standardised is more useful for
+  consumers but requires maintenance as endpoints evolve.
+
+- **Event ordering guarantees.**  Pub/sub does not guarantee
+  ordering across partitions.  Consumers that need strict ordering
+  must sort by `timestamp` or `request_id` sequence.
+
+## References
+
+- [IAM Contract](iam-contract.md) — the authentication/authorisation
+  abstraction.
+- [IAM Protocol](iam-protocol.md) — the OSS regime wire protocol.
+- [Capability Vocabulary](capabilities.md) — the capability strings
+  used in authorisation and audit events.
+- [Logging Strategy](logging-strategy.md) — operational logging
+  (complementary, not overlapping).

tests/contract/test_schema_field_contracts.py

@@ -24,6 +24,8 @@ from trustgraph.schema import (
     EntityContext,
     EntityEmbeddings,
     ChunkEmbeddings,
+    AuditEvent,
+    IamRequest,
 )
 
 
@@ -72,3 +74,21 @@ class TestSchemaFieldContracts:
             "context",
             "chunk_id",
         }
+
+    def test_audit_event_fields(self):
+        assert _field_names(AuditEvent) == {
+            "schema_version",
+            "event_id",
+            "event_type",
+            "timestamp",
+            "producer",
+            "payload_json",
+        }
+
+    def test_iam_request_has_audit_fields(self):
+        """IamRequest must carry request_id and client_ip for audit
+        correlation.  Removing these breaks the gateway→IAM audit
+        chain."""
+        names = _field_names(IamRequest)
+        assert "request_id" in names
+        assert "client_ip" in names

tests/unit/test_base/test_audit_publisher.py

@@ -0,0 +1,90 @@
+"""
+Tests for the AuditPublisher utility.
+
+Verifies envelope construction, fire-and-forget semantics, and
+failure suppression.
+"""
+
+import json
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from trustgraph.base.audit_publisher import AuditPublisher
+from trustgraph.schema import AuditEvent, audit_events_queue
+
+
+class TestAuditPublisherInit:
+
+    def test_queue_is_notify_class(self):
+        assert audit_events_queue == "notify:tg:audit-events"
+
+    def test_creates_producer_with_audit_queue(self):
+        backend = MagicMock()
+        pub = AuditPublisher(
+            backend=backend,
+            component_name="test-component",
+        )
+        assert pub.producer.topic == audit_events_queue
+        assert pub.producer.schema == AuditEvent
+        assert pub.component_name == "test-component"
+
+
+class TestAuditPublisherEmit:
+
+    @pytest.fixture
+    def publisher(self):
+        backend = MagicMock()
+        pub = AuditPublisher(
+            backend=backend,
+            component_name="test-svc",
+            processor_id="proc-1",
+        )
+        pub.producer = AsyncMock()
+        return pub
+
+    @pytest.mark.asyncio
+    async def test_emit_sends_structured_envelope(self, publisher):
+        await publisher.emit("gateway.request", {"path": "/test"})
+
+        publisher.producer.send.assert_called_once()
+        event = publisher.producer.send.call_args[0][0]
+
+        assert isinstance(event, AuditEvent)
+        assert event.schema_version == 1
+        assert event.event_type == "gateway.request"
+        assert event.producer == "test-svc"
+        assert event.event_id != ""
+        assert event.timestamp != ""
+
+    @pytest.mark.asyncio
+    async def test_emit_serializes_payload_as_json(self, publisher):
+        payload = {"method": "POST", "status_code": 200}
+        await publisher.emit("gateway.request", payload)
+
+        event = publisher.producer.send.call_args[0][0]
+        decoded = json.loads(event.payload_json)
+        assert decoded == payload
+
+    @pytest.mark.asyncio
+    async def test_emit_generates_unique_event_ids(self, publisher):
+        await publisher.emit("test.a", {})
+        await publisher.emit("test.b", {})
+
+        ids = [
+            call[0][0].event_id
+            for call in publisher.producer.send.call_args_list
+        ]
+        assert ids[0] != ids[1]
+
+    @pytest.mark.asyncio
+    async def test_emit_swallows_send_failure(self, publisher):
+        publisher.producer.send.side_effect = RuntimeError("pub/sub down")
+        await publisher.emit("test.event", {"key": "value"})
+
+    @pytest.mark.asyncio
+    async def test_emit_timestamp_is_utc_iso(self, publisher):
+        await publisher.emit("test.event", {})
+
+        event = publisher.producer.send.call_args[0][0]
+        assert "T" in event.timestamp
+        assert "+" in event.timestamp or "Z" in event.timestamp

tests/unit/test_gateway/test_audit_middleware.py

@@ -0,0 +1,198 @@
+"""
+Tests for the gateway audit middleware.
+
+Verifies that gateway.request events are emitted with correct
+metadata for success, error, and auth failure paths.
+"""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+from aiohttp import web
+from aiohttp.test_utils import make_mocked_request
+
+from trustgraph.gateway.audit import make_audit_middleware, _client_ip, _outcome_from_status
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_request(method="POST", path="/api/v1/config", headers=None,
+                  remote="10.0.0.1"):
+    hdrs = headers or {}
+    req = make_mocked_request(method, path, headers=hdrs)
+    req._transport_peername = (remote, 0)
+    return req
+
+
+# ---------------------------------------------------------------------------
+# Pure helpers
+# ---------------------------------------------------------------------------
+
+class TestClientIp:
+
+    def test_uses_x_forwarded_for(self):
+        req = _make_request(headers={"X-Forwarded-For": "1.2.3.4, 10.0.0.1"})
+        assert _client_ip(req) == "1.2.3.4"
+
+    def test_falls_back_to_remote(self):
+        req = _make_request(remote="192.168.1.1")
+        assert _client_ip(req) == "192.168.1.1"
+
+
+class TestOutcomeFromStatus:
+
+    def test_success_range(self):
+        assert _outcome_from_status(200) == "success"
+        assert _outcome_from_status(201) == "success"
+        assert _outcome_from_status(204) == "success"
+        assert _outcome_from_status(301) == "success"
+
+    def test_unauthenticated(self):
+        assert _outcome_from_status(401) == "unauthenticated"
+
+    def test_denied(self):
+        assert _outcome_from_status(403) == "denied"
+
+    def test_error(self):
+        assert _outcome_from_status(400) == "error"
+        assert _outcome_from_status(404) == "error"
+        assert _outcome_from_status(500) == "error"
+
+
+# ---------------------------------------------------------------------------
+# Middleware integration
+# ---------------------------------------------------------------------------
+
+class TestAuditMiddleware:
+
+    @pytest.fixture
+    def audit_publisher(self):
+        pub = AsyncMock()
+        pub.emit = AsyncMock()
+        return pub
+
+    @pytest.fixture
+    def middleware(self, audit_publisher):
+        return make_audit_middleware(audit_publisher)
+
+    @pytest.mark.asyncio
+    async def test_emits_event_on_success(self, middleware, audit_publisher):
+        async def handler(request):
+            return web.json_response({"ok": True})
+
+        req = _make_request()
+        await middleware(req, handler)
+
+        audit_publisher.emit.assert_called_once()
+        event_type, payload = audit_publisher.emit.call_args[0]
+        assert event_type == "gateway.request"
+        assert payload["method"] == "POST"
+        assert payload["path"] == "/api/v1/config"
+        assert payload["status_code"] == 200
+        assert payload["outcome"] == "success"
+        assert "request_id" in payload
+        assert "duration_ms" in payload
+        assert "error" not in payload
+
+    @pytest.mark.asyncio
+    async def test_emits_event_on_http_exception(self, middleware,
+                                                  audit_publisher):
+        async def handler(request):
+            raise web.HTTPForbidden(reason="access denied")
+
+        req = _make_request()
+        with pytest.raises(web.HTTPForbidden):
+            await middleware(req, handler)
+
+        audit_publisher.emit.assert_called_once()
+        _, payload = audit_publisher.emit.call_args[0]
+        assert payload["status_code"] == 403
+        assert payload["outcome"] == "denied"
+        assert payload["error"] == "access denied"
+
+    @pytest.mark.asyncio
+    async def test_emits_event_on_unhandled_exception(self, middleware,
+                                                       audit_publisher):
+        async def handler(request):
+            raise ValueError("boom")
+
+        req = _make_request()
+        with pytest.raises(ValueError):
+            await middleware(req, handler)
+
+        audit_publisher.emit.assert_called_once()
+        _, payload = audit_publisher.emit.call_args[0]
+        assert payload["status_code"] == 500
+        assert payload["outcome"] == "error"
+        assert payload["error"] == "ValueError"
+
+    @pytest.mark.asyncio
+    async def test_includes_identity_when_annotated(self, middleware,
+                                                     audit_publisher):
+        async def handler(request):
+            request['audit_identity'] = "user:mark"
+            return web.json_response({"ok": True})
+
+        req = _make_request()
+        await middleware(req, handler)
+
+        _, payload = audit_publisher.emit.call_args[0]
+        assert payload["identity"] == "user:mark"
+
+    @pytest.mark.asyncio
+    async def test_includes_capability_when_annotated(self, middleware,
+                                                       audit_publisher):
+        async def handler(request):
+            request['audit_capability'] = "config:read"
+            request['audit_workspace'] = "production"
+            return web.json_response({"ok": True})
+
+        req = _make_request()
+        await middleware(req, handler)
+
+        _, payload = audit_publisher.emit.call_args[0]
+        assert payload["capability"] == "config:read"
+        assert payload["workspace"] == "production"
+
+    @pytest.mark.asyncio
+    async def test_omits_unannotated_fields(self, middleware,
+                                             audit_publisher):
+        async def handler(request):
+            return web.json_response({"ok": True})
+
+        req = _make_request()
+        await middleware(req, handler)
+
+        _, payload = audit_publisher.emit.call_args[0]
+        assert "identity" not in payload
+        assert "capability" not in payload
+        assert "workspace" not in payload
+
+    @pytest.mark.asyncio
+    async def test_assigns_request_id(self, middleware, audit_publisher):
+        captured_id = None
+
+        async def handler(request):
+            nonlocal captured_id
+            captured_id = request.get('audit_request_id')
+            return web.json_response({"ok": True})
+
+        req = _make_request()
+        await middleware(req, handler)
+
+        assert captured_id is not None
+        _, payload = audit_publisher.emit.call_args[0]
+        assert payload["request_id"] == captured_id
+
+    @pytest.mark.asyncio
+    async def test_still_returns_response_if_emit_fails(self, middleware,
+                                                         audit_publisher):
+        audit_publisher.emit.side_effect = RuntimeError("pub/sub down")
+
+        async def handler(request):
+            return web.json_response({"ok": True})
+
+        req = _make_request()
+        resp = await middleware(req, handler)
+        assert resp.status == 200

tests/unit/test_gateway/test_auth.py

@@ -244,7 +244,7 @@ class TestIamAuthDispatch:
     async def test_api_key_path(self):
         auth = IamAuth(backend=Mock())
 
-        async def fake_resolve(api_key):
+        async def fake_resolve(api_key, **kwargs):
             assert api_key == "tg_testkey"
             # Roles are returned by the regime as a hint but the
             # gateway ignores them — kept here so the resolve
@@ -309,7 +309,7 @@ class TestApiKeyCache:
         seen = []
 
         async def fake_with_client(op):
-            async def resolve(plaintext):
+            async def resolve(plaintext, **kwargs):
                 seen.append(plaintext)
                 return ("u-" + plaintext, "default", ["reader"])
             return await op(Mock(resolve_api_key=resolve))

tests/unit/test_iam/test_iam_audit_events.py

@@ -0,0 +1,345 @@
+"""
+Tests for IAM service audit event emission.
+
+Verifies that the IAM Processor emits correctly categorised audit
+events (iam.authenticate, iam.authorise, iam.management) after
+handling each request type.
+"""
+
+import asyncio
+import pytest
+from unittest.mock import AsyncMock, MagicMock, Mock
+
+from trustgraph.schema import IamRequest, IamResponse, Error
+from trustgraph.iam.service.service import Processor
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_msg(request, msg_id="test-msg-1"):
+    msg = MagicMock()
+    msg.value.return_value = request
+    msg.properties.return_value = {"id": msg_id}
+    return msg
+
+
+def _make_processor():
+    """Create a Processor with stubbed dependencies."""
+    proc = object.__new__(Processor)
+    proc.id = "test-iam-svc"
+    proc.iam_response_producer = AsyncMock()
+    proc.audit = AsyncMock()
+    return proc
+
+
+# ---------------------------------------------------------------------------
+# Authentication events
+# ---------------------------------------------------------------------------
+
+class TestAuthenticateAuditEvents:
+
+    @pytest.mark.asyncio
+    async def test_resolve_api_key_success(self):
+        proc = _make_processor()
+        resp = IamResponse(
+            resolved_user_id="user-123",
+            resolved_default_workspace="default",
+        )
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="resolve-api-key",
+            api_key="tg_test",
+            request_id="req-1",
+            client_ip="10.0.0.1",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        proc.audit.emit.assert_called_once()
+        event_type, payload = proc.audit.emit.call_args[0]
+        assert event_type == "iam.authenticate"
+        assert payload["credential_type"] == "api-key"
+        assert payload["identity"] == "user-123"
+        assert payload["outcome"] == "success"
+        assert payload["request_id"] == "req-1"
+        assert payload["client_ip"] == "10.0.0.1"
+        assert "failure_reason" not in payload
+
+    @pytest.mark.asyncio
+    async def test_resolve_api_key_failure(self):
+        proc = _make_processor()
+        resp = IamResponse(error=Error(type="auth-failed", message="unknown"))
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="resolve-api-key",
+            api_key="tg_bad",
+            request_id="req-2",
+            client_ip="10.0.0.2",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        _, payload = proc.audit.emit.call_args[0]
+        assert payload["outcome"] == "failure"
+        assert payload["identity"] == "unknown"
+        assert payload["failure_reason"] == "auth-failed"
+
+    @pytest.mark.asyncio
+    async def test_login_emits_authenticate(self):
+        proc = _make_processor()
+        resp = IamResponse(
+            resolved_user_id="user-456",
+            jwt="eyJ...",
+            jwt_expires="2026-07-06T10:00:00Z",
+        )
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="login",
+            username="mark",
+            password="secret",
+            request_id="req-3",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        event_type, payload = proc.audit.emit.call_args[0]
+        assert event_type == "iam.authenticate"
+        assert payload["credential_type"] == "login-password"
+
+    @pytest.mark.asyncio
+    async def test_anonymous_emits_authenticate(self):
+        proc = _make_processor()
+        resp = IamResponse(error=Error(type="auth-failed", message="no"))
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(operation="authenticate-anonymous")
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        event_type, payload = proc.audit.emit.call_args[0]
+        assert event_type == "iam.authenticate"
+        assert payload["credential_type"] == "anonymous"
+        assert payload["outcome"] == "failure"
+
+
+# ---------------------------------------------------------------------------
+# Authorise events
+# ---------------------------------------------------------------------------
+
+class TestAuthoriseAuditEvents:
+
+    @pytest.mark.asyncio
+    async def test_authorise_allow(self):
+        proc = _make_processor()
+        resp = IamResponse(decision_allow=True, decision_ttl_seconds=60)
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="authorise",
+            user_id="user-1",
+            capability="config:read",
+            resource_json='{"workspace": "production"}',
+            request_id="req-4",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        event_type, payload = proc.audit.emit.call_args[0]
+        assert event_type == "iam.authorise"
+        assert payload["outcome"] == "allow"
+        assert payload["capability"] == "config:read"
+        assert payload["identity"] == "user-1"
+        assert payload["workspace"] == "production"
+        assert "denial_reason" not in payload
+
+    @pytest.mark.asyncio
+    async def test_authorise_deny(self):
+        proc = _make_processor()
+        resp = IamResponse(decision_allow=False, decision_ttl_seconds=5)
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="authorise",
+            user_id="user-2",
+            capability="users:admin",
+            resource_json='{}',
+            request_id="req-5",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        _, payload = proc.audit.emit.call_args[0]
+        assert payload["outcome"] == "deny"
+        assert payload["denial_reason"] == "capability-not-in-role"
+
+    @pytest.mark.asyncio
+    async def test_authorise_extracts_workspace_from_resource_json(self):
+        proc = _make_processor()
+        resp = IamResponse(decision_allow=True, decision_ttl_seconds=60)
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="authorise",
+            user_id="user-1",
+            capability="graph-rag:query",
+            resource_json='{"workspace": "engineering"}',
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        _, payload = proc.audit.emit.call_args[0]
+        assert payload["workspace"] == "engineering"
+
+    @pytest.mark.asyncio
+    async def test_authorise_omits_empty_workspace(self):
+        proc = _make_processor()
+        resp = IamResponse(decision_allow=True, decision_ttl_seconds=60)
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="authorise",
+            user_id="user-1",
+            capability="metrics:read",
+            resource_json='{}',
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        _, payload = proc.audit.emit.call_args[0]
+        assert "workspace" not in payload
+
+
+# ---------------------------------------------------------------------------
+# Management events
+# ---------------------------------------------------------------------------
+
+class TestManagementAuditEvents:
+
+    @pytest.mark.asyncio
+    async def test_create_user_emits_management(self):
+        proc = _make_processor()
+        resp = IamResponse()
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="create-user",
+            actor="admin-1",
+            user_id="new-user",
+            workspace="default",
+            request_id="req-6",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        event_type, payload = proc.audit.emit.call_args[0]
+        assert event_type == "iam.management"
+        assert payload["operation"] == "create-user"
+        assert payload["actor"] == "admin-1"
+        assert payload["target_identity"] == "new-user"
+        assert payload["target_workspace"] == "default"
+        assert payload["outcome"] == "success"
+
+    @pytest.mark.asyncio
+    async def test_revoke_api_key_emits_management(self):
+        proc = _make_processor()
+        resp = IamResponse()
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="revoke-api-key",
+            actor="admin-1",
+            key_id="key-abc",
+            workspace="production",
+            request_id="req-7",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        event_type, payload = proc.audit.emit.call_args[0]
+        assert event_type == "iam.management"
+        assert payload["operation"] == "revoke-api-key"
+        assert payload["outcome"] == "success"
+
+    @pytest.mark.asyncio
+    async def test_management_error_outcome(self):
+        proc = _make_processor()
+        resp = IamResponse(
+            error=Error(type="not-found", message="user not found"),
+        )
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="delete-user",
+            actor="admin-1",
+            user_id="ghost",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        _, payload = proc.audit.emit.call_args[0]
+        assert payload["outcome"] == "error"
+
+    @pytest.mark.asyncio
+    async def test_management_omits_empty_target_fields(self):
+        proc = _make_processor()
+        resp = IamResponse()
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(
+            operation="rotate-signing-key",
+            actor="admin-1",
+        )
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        _, payload = proc.audit.emit.call_args[0]
+        assert "target_identity" not in payload
+        assert "target_workspace" not in payload
+
+
+# ---------------------------------------------------------------------------
+# Non-audited operations
+# ---------------------------------------------------------------------------
+
+class TestNonAuditedOperations:
+
+    @pytest.mark.asyncio
+    async def test_whoami_does_not_emit(self):
+        proc = _make_processor()
+        resp = IamResponse()
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(operation="whoami", actor="user-1")
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        proc.audit.emit.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_list_users_does_not_emit(self):
+        proc = _make_processor()
+        resp = IamResponse()
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(operation="list-users", workspace="default")
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        proc.audit.emit.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_get_signing_key_does_not_emit(self):
+        proc = _make_processor()
+        resp = IamResponse(signing_key_public="PEM...")
+        proc.iam = MagicMock()
+        proc.iam.handle = AsyncMock(return_value=resp)
+
+        req = IamRequest(operation="get-signing-key-public")
+        await proc.on_iam_request(_make_msg(req), None, None)
+
+        proc.audit.emit.assert_not_called()

trustgraph-base/trustgraph/base/__init__.py

@@ -46,4 +46,5 @@ from . reranker_client import RerankerClientSpec
 from . reranker_service import RerankerService
 from . row_embeddings_query_client import RowEmbeddingsQueryClientSpec
 from . collection_config_handler import CollectionConfigHandler
+from . audit_publisher import AuditPublisher
 

trustgraph-base/trustgraph/base/audit_publisher.py

@@ -0,0 +1,42 @@
+
+import json
+import logging
+from datetime import datetime, timezone
+from uuid import uuid4
+
+from . producer import Producer
+from . metrics import ProducerMetrics
+from trustgraph.schema import AuditEvent, audit_events_queue
+
+logger = logging.getLogger(__name__)
+
+
+class AuditPublisher:
+
+    def __init__(self, backend, component_name, processor_id=None):
+        self.component_name = component_name
+        self.producer = Producer(
+            backend=backend,
+            topic=audit_events_queue,
+            schema=AuditEvent,
+            metrics=ProducerMetrics(
+                processor=processor_id or component_name,
+                flow=None,
+                name="audit-events",
+            ),
+        )
+
+    async def emit(self, event_type, payload):
+        event = AuditEvent(
+            schema_version=1,
+            event_id=str(uuid4()),
+            event_type=event_type,
+            timestamp=datetime.now(timezone.utc).isoformat(),
+            producer=self.component_name,
+            payload_json=json.dumps(payload),
+        )
+
+        try:
+            await self.producer.send(event)
+        except Exception as e:
+            logger.warning(f"Failed to emit audit event: {e}")

trustgraph-base/trustgraph/base/iam_client.py

@@ -62,7 +62,8 @@ class IamClient(RequestResponse):
         )
         return resp.user
 
-    async def authenticate_anonymous(self, timeout=IAM_TIMEOUT):
+    async def authenticate_anonymous(self, timeout=IAM_TIMEOUT,
+                                     request_id="", client_ip=""):
         """Request anonymous access from the IAM regime.
 
         Returns ``(user_id, default_workspace, roles)`` if the regime
@@ -70,6 +71,8 @@ class IamClient(RequestResponse):
         error type ``auth-failed`` if it does not."""
         resp = await self._request(
             operation="authenticate-anonymous",
+            request_id=request_id,
+            client_ip=client_ip,
             timeout=timeout,
         )
         return (
@@ -78,7 +81,8 @@ class IamClient(RequestResponse):
             list(resp.resolved_roles),
         )
 
-    async def resolve_api_key(self, api_key, timeout=IAM_TIMEOUT):
+    async def resolve_api_key(self, api_key, timeout=IAM_TIMEOUT,
+                              request_id="", client_ip=""):
         """Resolve a plaintext API key to its identity triple.
 
         Returns ``(user_id, default_workspace, roles)`` or raises
@@ -87,6 +91,8 @@ class IamClient(RequestResponse):
         resp = await self._request(
             operation="resolve-api-key",
             api_key=api_key,
+            request_id=request_id,
+            client_ip=client_ip,
             timeout=timeout,
         )
         return (
@@ -96,7 +102,8 @@ class IamClient(RequestResponse):
         )
 
     async def authorise(self, identity_handle, capability,
-                        resource, parameters, timeout=IAM_TIMEOUT):
+                        resource, parameters, timeout=IAM_TIMEOUT,
+                        request_id="", client_ip=""):
         """Ask the IAM regime whether ``identity_handle`` may perform
         ``capability`` on ``resource`` given ``parameters``.
 
@@ -111,6 +118,8 @@ class IamClient(RequestResponse):
             capability=capability,
             resource_json=json.dumps(resource or {}, sort_keys=True),
             parameters_json=json.dumps(parameters or {}, sort_keys=True),
+            request_id=request_id,
+            client_ip=client_ip,
             timeout=timeout,
         )
         return resp.decision_allow, resp.decision_ttl_seconds
@@ -186,7 +195,7 @@ class IamClient(RequestResponse):
         )
 
     async def login(self, username, password, workspace="",
-                    timeout=IAM_TIMEOUT):
+                    timeout=IAM_TIMEOUT, request_id="", client_ip=""):
         """Validate credentials and return ``(jwt, expires_iso)``.
         ``workspace`` is optional; defaults at the server to the
         OSS default workspace."""
@@ -195,6 +204,8 @@ class IamClient(RequestResponse):
             workspace=workspace,
             username=username,
             password=password,
+            request_id=request_id,
+            client_ip=client_ip,
             timeout=timeout,
         )
         return resp.jwt, resp.jwt_expires

trustgraph-base/trustgraph/schema/services/__init__.py

@@ -16,4 +16,5 @@ from .collection import *
 from .storage import *
 from .tool_service import *
 from .sparql_query import *
-from .reranker import *
+from .reranker import *
+from .audit import *

trustgraph-base/trustgraph/schema/services/audit.py

@@ -0,0 +1,27 @@
+
+from dataclasses import dataclass, field
+
+from ..core.topic import queue
+
+############################################################################
+
+# Audit events — see docs/tech-specs/audit-events.md for the full spec.
+#
+# Transport: notify-class pub/sub (fire-and-forget, per-subscriber
+# delivery).  Producers are the API gateway and the IAM service.
+# Consumers are optional enterprise components.
+
+
+@dataclass
+class AuditEvent:
+    schema_version: int = 1
+    event_id: str = ""
+    event_type: str = ""
+    timestamp: str = ""
+    producer: str = ""
+    payload_json: str = ""
+
+
+audit_events_queue = queue('audit-events', cls='notify')
+
+############################################################################

trustgraph-base/trustgraph/schema/services/iam.py

@@ -121,6 +121,10 @@ class IamRequest:
     group: GroupInput | None = None
     grant: GrantInput | None = None
 
+    # ---- Audit context (informational, echoed into audit events) ----
+    request_id: str = ""
+    client_ip: str = ""
+
     # ---- authorise / authorise-many inputs ----
     # Capability string from the vocabulary in capabilities.md.
     capability: str = ""

trustgraph-flow/trustgraph/gateway/audit.py

@@ -0,0 +1,105 @@
+"""
+Audit middleware for the API gateway.
+
+Wraps every HTTP request with timing and metadata collection, then
+emits a ``gateway.request`` audit event after the response is sent.
+Handlers can enrich the event by annotating the request dict:
+
+    request['audit_identity'] = identity.principal_id
+    request['audit_capability'] = capability
+    request['audit_workspace'] = workspace
+"""
+
+import time
+import logging
+from uuid import uuid4
+
+from aiohttp import web
+
+from trustgraph.base import AuditPublisher
+
+logger = logging.getLogger("audit")
+
+
+def _client_ip(request):
+    forwarded = request.headers.get("X-Forwarded-For")
+    if forwarded:
+        return forwarded.split(",")[0].strip()
+    peer = request.remote
+    return peer or ""
+
+
+def _outcome_from_status(status):
+    if 200 <= status < 400:
+        return "success"
+    if status == 401:
+        return "unauthenticated"
+    if status == 403:
+        return "denied"
+    return "error"
+
+
+def make_audit_middleware(audit_publisher):
+
+    @web.middleware
+    async def audit_middleware(request, handler):
+        request_id = str(uuid4())
+        request['audit_request_id'] = request_id
+        start = time.monotonic()
+
+        status_code = 500
+        error = None
+        response_size = 0
+
+        try:
+            response = await handler(request)
+            status_code = response.status
+            response_size = response.content_length or 0
+            return response
+        except web.HTTPException as exc:
+            status_code = exc.status
+            error = exc.reason
+            raise
+        except Exception as exc:
+            status_code = 500
+            error = type(exc).__name__
+            raise
+        finally:
+            duration_ms = int((time.monotonic() - start) * 1000)
+            outcome = _outcome_from_status(status_code)
+
+            payload = {
+                "request_id": request_id,
+                "method": request.method,
+                "path": request.path,
+                "client_ip": _client_ip(request),
+                "user_agent": request.headers.get("User-Agent", ""),
+                "status_code": status_code,
+                "outcome": outcome,
+                "duration_ms": duration_ms,
+                "request_size_bytes": request.content_length or 0,
+                "response_size_bytes": response_size,
+            }
+
+            identity = request.get('audit_identity')
+            if identity:
+                payload["identity"] = identity
+
+            capability = request.get('audit_capability')
+            if capability:
+                payload["capability"] = capability
+
+            workspace = request.get('audit_workspace')
+            if workspace:
+                payload["workspace"] = workspace
+
+            if error and outcome != "success":
+                payload["error"] = error
+
+            try:
+                await audit_publisher.emit("gateway.request", payload)
+            except Exception:
+                logger.debug("Failed to emit gateway audit event",
+                             exc_info=True)
+
+    return audit_middleware

trustgraph-flow/trustgraph/gateway/auth.py

@@ -232,21 +232,55 @@ class IamAuth:
         cannot distinguish missing / malformed / invalid / expired /
         revoked credentials."""
 
+        request_id = request.get('audit_request_id', '') if hasattr(request, 'get') else ''
+        client_ip = self._extract_client_ip(request)
+
         header = request.headers.get("Authorization", "")
         if not header.startswith("Bearer "):
-            return await self._authenticate_anonymous()
+            identity = await self._authenticate_anonymous(
+                request_id=request_id, client_ip=client_ip,
+            )
+            self._annotate_request(request, identity)
+            return identity
         token = header[len("Bearer "):].strip()
         if not token:
-            return await self._authenticate_anonymous()
+            identity = await self._authenticate_anonymous(
+                request_id=request_id, client_ip=client_ip,
+            )
+            self._annotate_request(request, identity)
+            return identity
 
         # API keys always start with "tg_".  JWTs have two dots and
         # no "tg_" prefix.  Discriminate cheaply.
         if token.startswith("tg_"):
-            return await self._resolve_api_key(token)
+            identity = await self._resolve_api_key(
+                token, request_id=request_id, client_ip=client_ip,
+            )
+            self._annotate_request(request, identity)
+            return identity
         if token.count(".") == 2:
-            return self._verify_jwt(token)
+            identity = self._verify_jwt(token)
+            self._annotate_request(request, identity)
+            return identity
         raise _auth_failure()
 
+    @staticmethod
+    def _extract_client_ip(request):
+        try:
+            forwarded = request.headers.get("X-Forwarded-For")
+            if forwarded:
+                return forwarded.split(",")[0].strip()
+            return request.remote or ""
+        except Exception:
+            return ""
+
+    @staticmethod
+    def _annotate_request(request, identity):
+        try:
+            request['audit_identity'] = identity.principal_id
+        except Exception:
+            pass
+
     def _verify_jwt(self, token):
         if not self._signing_public_pem:
             raise _auth_failure()
@@ -266,10 +300,12 @@ class IamAuth:
             principal_id=sub, source="jwt",
         )
 
-    async def _authenticate_anonymous(self):
+    async def _authenticate_anonymous(self, request_id="", client_ip=""):
         try:
             async def _call(client):
-                return await client.authenticate_anonymous()
+                return await client.authenticate_anonymous(
+                    request_id=request_id, client_ip=client_ip,
+                )
             user_id, default_workspace, _roles = await self._with_client(
                 _call,
             )
@@ -288,7 +324,7 @@ class IamAuth:
             principal_id=user_id, source="anonymous",
         )
 
-    async def _resolve_api_key(self, plaintext):
+    async def _resolve_api_key(self, plaintext, request_id="", client_ip=""):
         h = hashlib.sha256(plaintext.encode("utf-8")).hexdigest()
 
         cached = self._key_cache.get(h)
@@ -303,7 +339,10 @@ class IamAuth:
 
             try:
                 async def _call(client):
-                    return await client.resolve_api_key(plaintext)
+                    return await client.resolve_api_key(
+                        plaintext, request_id=request_id,
+                        client_ip=client_ip,
+                    )
                 # ``roles`` is returned by the OSS regime as a hint
                 # but is not consulted by the gateway; all policy
                 # decisions go through ``authorise``.
@@ -345,7 +384,8 @@ class IamAuth:
         )
         return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
-    async def authorise(self, identity, capability, resource, parameters):
+    async def authorise(self, identity, capability, resource, parameters,
+                        request_id="", client_ip=""):
         """Ask the IAM regime whether ``identity`` may perform
         ``capability`` on ``resource`` given ``parameters``.
 
@@ -383,6 +423,7 @@ class IamAuth:
                     return await client.authorise(
                         identity.handle, capability,
                         resource or {}, parameters or {},
+                        request_id=request_id, client_ip=client_ip,
                     )
                 allow, ttl = await self._with_client(_call)
             except Exception as e:

trustgraph-flow/trustgraph/gateway/endpoint/iam_endpoint.py

@@ -90,7 +90,13 @@ class IamEndpoint:
 
             await self.auth.authorise(
                 identity, op.capability, resource, parameters,
+                request_id=request.get('audit_request_id', ''),
+                client_ip=self.auth._extract_client_ip(request),
             )
+            request['audit_capability'] = op.capability
+            ws = resource.get('workspace', '')
+            if ws:
+                request['audit_workspace'] = ws
 
         # Plumb the authenticated caller's handle through as ``actor``
         # so iam-svc handlers (e.g. whoami, future actor-scoped

trustgraph-flow/trustgraph/gateway/endpoint/manager.py

@@ -75,9 +75,14 @@ class _RoutedVariableEndpoint:
             parameters = op.extract_parameters(ctx)
             await self.auth.authorise(
                 identity, op.capability, resource, parameters,
+                request_id=request.get('audit_request_id', ''),
+                client_ip=self.auth._extract_client_ip(request),
             )
+            request['audit_capability'] = op.capability
 
             ws = resource.get("workspace", "")
+            if ws:
+                request['audit_workspace'] = ws
             if ws and ws not in self.auth.known_workspaces:
                 raise workspace_not_found()
 
@@ -143,9 +148,14 @@ class _RoutedSocketEndpoint:
             parameters = op.extract_parameters(ctx)
             await self.auth.authorise(
                 identity, op.capability, resource, parameters,
+                request_id=request.get('audit_request_id', ''),
+                client_ip=self.auth._extract_client_ip(request),
             )
+            request['audit_capability'] = op.capability
 
             ws = resource.get("workspace", "")
+            if ws:
+                request['audit_workspace'] = ws
             if ws and ws not in self.auth.known_workspaces:
                 raise workspace_not_found()
 

trustgraph-flow/trustgraph/gateway/endpoint/registry_endpoint.py

@@ -98,7 +98,13 @@ class RegistryRoutedVariableEndpoint:
 
             await self.auth.authorise(
                 identity, op.capability, resource, parameters,
+                request_id=request.get('audit_request_id', ''),
+                client_ip=self.auth._extract_client_ip(request),
             )
+            request['audit_capability'] = op.capability
+            ws = resource.get('workspace', '')
+            if ws:
+                request['audit_workspace'] = ws
 
             # Default-fill workspace into the body so downstream
             # dispatchers see the canonical resolved value.  The

trustgraph-flow/trustgraph/gateway/service.py

@@ -11,8 +11,10 @@ import os
 
 from trustgraph.base.logging import setup_logging, add_logging_args
 from trustgraph.base.pubsub import get_pubsub, add_pubsub_args
+from trustgraph.base import AuditPublisher
 
 from . auth import IamAuth
+from . audit import make_audit_middleware
 from . config.receiver import ConfigReceiver
 from . dispatch.manager import DispatcherManager
 
@@ -122,12 +124,18 @@ class Api:
             timeout = self.timeout,
         )
 
+        self.audit_publisher = AuditPublisher(
+            backend=self.pubsub_backend,
+            component_name="api-gateway",
+            processor_id=config.get("id", "api-gateway"),
+        )
+
         self.endpoint_manager = EndpointManager(
             dispatcher_manager = self.dispatcher_manager,
             auth = self.auth,
             prometheus_url = self.prometheus_url,
             timeout = self.timeout,
-            
+
         )
 
         self.endpoints = [
@@ -136,7 +144,7 @@ class Api:
     async def app_factory(self):
 
         self.app = web.Application(
-            middlewares=[],
+            middlewares=[make_audit_middleware(self.audit_publisher)],
             client_max_size=256 * 1024 * 1024
         )
 

trustgraph-flow/trustgraph/iam/service/service.py

@@ -16,6 +16,7 @@ from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigValue
 from trustgraph.schema import config_request_queue, config_response_queue
 
 from trustgraph.base import AsyncProcessor, Consumer, Producer
+from trustgraph.base import AuditPublisher
 from trustgraph.base import ConsumerMetrics, ProducerMetrics
 from trustgraph.base.metrics import SubscriberMetrics
 from trustgraph.base.request_response_spec import RequestResponse
@@ -145,6 +146,12 @@ class Processor(AsyncProcessor):
             metrics=iam_response_metrics,
         )
 
+        self.audit = AuditPublisher(
+            backend=self.pubsub,
+            component_name="iam-service",
+            processor_id=self.id,
+        )
+
         self.iam = IamService(
             host=self.cassandra_host,
             username=self.cassandra_username,
@@ -243,6 +250,19 @@ class Processor(AsyncProcessor):
                 f"{workspace_id}: {e}", exc_info=True,
             )
 
+    AUTHENTICATE_OPS = frozenset({
+        "resolve-api-key", "login", "authenticate-anonymous",
+    })
+    AUTHORISE_OPS = frozenset({
+        "authorise", "authorise-many",
+    })
+    MANAGEMENT_OPS = frozenset({
+        "create-user", "update-user", "disable-user", "enable-user",
+        "delete-user", "create-api-key", "revoke-api-key",
+        "create-workspace", "update-workspace", "disable-workspace",
+        "reset-password", "rotate-signing-key", "bootstrap",
+    })
+
     async def on_iam_request(self, msg, consumer, flow):
 
         id = None
@@ -256,6 +276,7 @@ class Processor(AsyncProcessor):
             await self.iam_response_producer.send(
                 resp, properties={"id": id},
             )
+            await self._emit_audit(v, resp)
         except Exception as e:
             logger.error(
                 f"IAM request failed: {type(e).__name__}: {e}",
@@ -269,6 +290,76 @@ class Processor(AsyncProcessor):
                     resp, properties={"id": id},
                 )
 
+    async def _emit_audit(self, v, resp):
+        try:
+            op = v.operation
+            if op in self.AUTHENTICATE_OPS:
+                await self._emit_authenticate(v, resp)
+            elif op in self.AUTHORISE_OPS:
+                await self._emit_authorise(v, resp)
+            elif op in self.MANAGEMENT_OPS:
+                await self._emit_management(v, resp)
+        except Exception:
+            logger.debug("Failed to emit IAM audit event", exc_info=True)
+
+    async def _emit_authenticate(self, v, resp):
+        has_error = resp.error is not None
+        payload = {
+            "request_id": v.request_id,
+            "credential_type": self._credential_type(v.operation),
+            "identity": resp.resolved_user_id if not has_error else "unknown",
+            "outcome": "failure" if has_error else "success",
+            "client_ip": v.client_ip,
+        }
+        if has_error:
+            payload["failure_reason"] = resp.error.type
+        if v.key_id:
+            payload["key_id"] = v.key_id
+        await self.audit.emit("iam.authenticate", payload)
+
+    async def _emit_authorise(self, v, resp):
+        import json as _json
+        workspace = v.workspace
+        if not workspace:
+            try:
+                resource = _json.loads(v.resource_json or "{}")
+                workspace = resource.get("workspace", "")
+            except Exception:
+                pass
+        payload = {
+            "request_id": v.request_id,
+            "identity": v.user_id,
+            "capability": v.capability,
+            "outcome": "allow" if resp.decision_allow else "deny",
+        }
+        if workspace:
+            payload["workspace"] = workspace
+        if not resp.decision_allow:
+            payload["denial_reason"] = "capability-not-in-role"
+        await self.audit.emit("iam.authorise", payload)
+
+    async def _emit_management(self, v, resp):
+        has_error = resp.error is not None
+        payload = {
+            "request_id": v.request_id,
+            "actor": v.actor,
+            "operation": v.operation,
+            "outcome": "error" if has_error else "success",
+        }
+        if v.user_id:
+            payload["target_identity"] = v.user_id
+        if v.workspace:
+            payload["target_workspace"] = v.workspace
+        await self.audit.emit("iam.management", payload)
+
+    @staticmethod
+    def _credential_type(operation):
+        if operation == "resolve-api-key":
+            return "api-key"
+        if operation == "login":
+            return "login-password"
+        return "anonymous"
+
     @staticmethod
     def add_args(parser):
         AsyncProcessor.add_args(parser)