Correct the audit-core envelope to the published wire contract

audit-core registered our sender (AUDIT-IN-0002) and, reviewing the
emitter, found that no event we sent could ever have been accepted.
envelope_for sent five of the eight required fields under its own names
-- event_id, action, resource, observed_at, details -- and omitted
correlation_id entirely. normalize() rejects that whole, 400.

Our drain treated 400 as terminal, so every event would have been marked
handled here while audit-core held only an unchained dead letter: lost on
both sides, silently, with the integration looking healthy.

- envelope_for emits exactly the eight required fields and none of the
  six audit-core derives. The acting principal moves into `data`, where
  it reads as our claim rather than the archive's finding.
- Thread correlation_id through create / revoke / plan, which had no such
  field. Optional on those three bodies for compatibility; when a caller
  supplies none this engine mints req-<uuid> for the operation it
  performed and returns it. The store mints op-<uuid> as a floor for
  direct callers, written into the local payload so both records agree.
- Send Idempotency-Key equal to the body id.
- A 400 no longer dead-letters. The row stays pending with the reason
  recorded on it: a 400 is an integration defect to fix, not a delivery
  outcome to record.
- wire_envelope upgrades outbox rows written in the old shape at send
  time, and refuses to send one whose correlation cannot be recovered
  from its payload rather than inventing one.

Verified by running all nine event types, through the API, through
audit-core's actual normalize() with a matching SenderIdentity -- all
accepted. A live non-production 202 still needs the token, so
TEN-WP-0012-T01 stays `wait`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DFmHM6fugwfqoobUCp9GiQ

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 2106375@bnt-lap001
Assistant-Session: aa26c34d-71e8-4478-a962-c79c74694dc8

src/tenant_engine/app.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import hashlib
 import json
 from datetime import UTC, datetime
+from uuid import uuid4
 
 from fastapi import FastAPI, Header, HTTPException, Query, Request, Response
 from fastapi.responses import JSONResponse
@@ -64,6 +65,11 @@ class CreateTenantRequest(BaseModel):
     actor: str
     display_name: str | None = None
     contact_email: str | None = None
+    # Optional, unlike the lifecycle bodies, because these three routes
+    # shipped without it and existing clients must keep working. When a
+    # caller has no correlation of its own this service mints one for the
+    # request and returns it -- see `_correlation`.
+    correlation_id: str | None = None
 
 
 class GrantRoleRequest(BaseModel):
@@ -79,11 +85,13 @@ class GrantRoleRequest(BaseModel):
 class RevokeRoleRequest(BaseModel):
     grant_id: str
     actor: str
+    correlation_id: str | None = None
 
 
 class AssignPlanRequest(BaseModel):
     plan_id: str
     actor: str
+    correlation_id: str | None = None
 
 
 class TenantMetadata(BaseModel):
@@ -277,6 +285,7 @@ def create_app(
 
     @app.post("/tenants", status_code=201)
     async def create_tenant(payload: CreateTenantRequest) -> dict:
+        correlation_id = _correlation(payload.correlation_id)
         outcome = _authorize(
             authorizer,
             store,
@@ -292,7 +301,9 @@ def create_app(
                 contact_email=payload.contact_email,
                 created_at=datetime.now(UTC),
             )
-            store.create_tenant(tenant, authz=outcome.as_payload())
+            store.create_tenant(
+                tenant, authz=outcome.as_payload(), correlation_id=correlation_id
+            )
             _drain_outbox(store, app.state.audit_core)
         except InvalidTenantIdentifierError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
@@ -300,7 +311,7 @@ def create_app(
             raise HTTPException(status_code=409, detail=str(exc)) from exc
         # Superset of the pre-lifecycle response (tenant_id/identifier/grouping
         # are unchanged), so existing create clients keep working.
-        return _tenant_response(tenant)
+        return {**_tenant_response(tenant), "correlation_id": correlation_id}
 
     @app.post("/tenants/{tenant_id}/roles/grant", status_code=201)
     async def grant_role(tenant_id: str, payload: GrantRoleRequest) -> dict:
@@ -327,10 +338,16 @@ def create_app(
             raise HTTPException(status_code=409, detail="tenant_retired") from exc
         except InvalidGrantError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
-        return {"grant_id": grant.grant_id, "tenant_id": tenant_id, "role": grant.role.value}
+        return {
+            "grant_id": grant.grant_id,
+            "tenant_id": tenant_id,
+            "role": grant.role.value,
+            "correlation_id": payload.correlation_id,
+        }
 
     @app.post("/tenants/{tenant_id}/roles/revoke")
     async def revoke_role(tenant_id: str, payload: RevokeRoleRequest) -> dict:
+        correlation_id = _correlation(payload.correlation_id)
         outcome = _authorize(
             authorizer, store, action="tenant.role.revoke", tenant_id=tenant_id, actor=payload.actor
         )
@@ -340,6 +357,7 @@ def create_app(
                 grant_id=payload.grant_id,
                 at=datetime.now(UTC),
                 authz=outcome.as_payload(),
+                correlation_id=correlation_id,
             )
             _drain_outbox(store, app.state.audit_core)
         except TenantNotFoundError as exc:
@@ -348,10 +366,16 @@ def create_app(
             raise HTTPException(status_code=404, detail="grant_not_found") from exc
         except InvalidGrantError as exc:
             raise HTTPException(status_code=409, detail=str(exc)) from exc
-        return {"grant_id": revoked.grant_id, "tenant_id": tenant_id, "revoked": True}
+        return {
+            "grant_id": revoked.grant_id,
+            "tenant_id": tenant_id,
+            "revoked": True,
+            "correlation_id": correlation_id,
+        }
 
     @app.post("/tenants/{tenant_id}/plan")
     async def assign_plan(tenant_id: str, payload: AssignPlanRequest) -> dict:
+        correlation_id = _correlation(payload.correlation_id)
         outcome = _authorize(
             authorizer, store, action="tenant.plan.assign", tenant_id=tenant_id, actor=payload.actor
         )
@@ -361,13 +385,18 @@ def create_app(
                     tenant_id=tenant_id, plan_id=payload.plan_id, assigned_at=datetime.now(UTC)
                 ),
                 authz=outcome.as_payload(),
+                correlation_id=correlation_id,
             )
             _drain_outbox(store, app.state.audit_core)
         except TenantNotFoundError as exc:
             raise HTTPException(status_code=404, detail="tenant_not_found") from exc
         except TenantRetiredError as exc:
             raise HTTPException(status_code=409, detail="tenant_retired") from exc
-        return {"tenant_id": tenant_id, "plan_id": payload.plan_id}
+        return {
+            "tenant_id": tenant_id,
+            "plan_id": payload.plan_id,
+            "correlation_id": correlation_id,
+        }
 
     # -- Lifecycle write API (TEN-WP-0005) --------------------------------
     # Update, retire, and reactivate are separately authorized actions, not
@@ -881,6 +910,18 @@ def _persist_authz(store: TenantStore, outcome: AuthorizationOutcome) -> None:
     )
 
 
+def _correlation(supplied: str | None) -> str:
+    """The correlation id for one mutation.
+
+    audit-core requires a truthy `correlation_id` on every event and will not
+    synthesize one, correctly: an id the archive invented would tie an event
+    to an operation nobody observed. This service *did* observe the operation,
+    so when a caller supplies none it mints an id for this request and returns
+    it in the response, where the caller can use it to correlate.
+    """
+    return supplied or f"req-{uuid4()}"
+
+
 def _drain_outbox(store: TenantStore, client: AuditCoreClient | None) -> None:
     """Best-effort drain. Never fails the mutation (attributive, non-blocking)."""
     if client is None:

src/tenant_engine/audit_core.py

@@ -19,42 +19,129 @@ from uuid import uuid4
 
 import httpx
 
-SCHEMA_VERSION = "audit-core.event.v1alpha1"
 SOURCE = "tenant-engine"
 
+#: The eight fields `POST /v1/events` requires, all of which must be truthy.
+#: Contract: audit-core `docs/event-envelope.md`. There is no version
+#: negotiation at that receiver -- an envelope either matches or is rejected
+#: whole -- so this tuple is the contract, not a schema string.
+REQUIRED_FIELDS = (
+    "id",
+    "type",
+    "source",
+    "subject",
+    "tenant",
+    "correlation_id",
+    "occurred_at",
+    "data",
+)
+
+
+class EnvelopeError(ValueError):
+    """An envelope this engine would not be able to deliver.
+
+    Raised at emission, inside the mutation transaction, rather than at
+    drain time. A malformed envelope is a defect in this service, and the
+    place to surface it is the mutation that produced it -- not the audit
+    trail, where audit-core would reject it into a dead-letter queue that
+    is a diagnostic surface and not custody.
+    """
+
 
 def new_event_id() -> str:
     return str(uuid4())
 
 
+def correlation_for(payload: dict[str, Any]) -> str:
+    """The correlation id for an event, minting one only if none was supplied.
+
+    audit-core requires a truthy `correlation_id` and will not synthesize one,
+    correctly: an id the *archive* invented would tie an event to an operation
+    it never observed. This engine is in a different position -- it performed
+    the operation -- so when a caller supplies no correlation it names its own
+    operation. The minted id is written back into the local event payload too,
+    so the local record and the external one carry the same id rather than
+    disagreeing about what the event belongs to.
+    """
+    return str(payload.get("correlation_id") or "") or f"op-{uuid4()}"
+
+
 def envelope_for(
     *,
     event_id: str,
     event_type: str,
     tenant_id: str,
-    observed_at: str,
+    occurred_at: str,
+    correlation_id: str,
     payload: dict[str, Any],
 ) -> dict[str, Any]:
-    return {
-        "schema_version": SCHEMA_VERSION,
-        "event_id": event_id,
-        "observed_at": observed_at,
-        "tenant": tenant_id,
-        "scope": "tenant-engine",
+    """Build the audit-core ingest envelope for one mutation event.
+
+    Only the fields the receiver reads are sent. `observed_at`, `action`,
+    `resource`, `scope`, `outcome`, and `actor` are derived by audit-core
+    from these; sending them would suggest this engine controls values it
+    does not. The acting principal travels inside `data`, where it reads as
+    this service's claim rather than the archive's finding.
+    """
+    envelope = {
+        "id": event_id,
+        "type": event_type,
         "source": SOURCE,
-        "actor": payload.get("actor") or payload.get("granted_by") or payload.get("changed_by"),
-        "action": event_type,
-        "resource": f"tenant:{tenant_id}",
-        "outcome": "recorded",
-        "reason": payload.get("reason") or payload.get("authorization_reason"),
-        "details": payload,
+        "subject": f"tenant:{tenant_id}",
+        "tenant": tenant_id,
+        "correlation_id": correlation_id,
+        "occurred_at": occurred_at,
+        "data": payload,
     }
+    missing = [field for field in REQUIRED_FIELDS if not envelope.get(field)]
+    if missing:
+        raise EnvelopeError(f"envelope missing required field(s): {', '.join(missing)}")
+    return envelope
+
+
+#: Envelopes written before the eight-field contract was published are still
+#: sitting in outboxes. Map the old key onto the new one at send time so a
+#: pending row from before the fix is deliverable rather than permanently
+#: rejected. `schema_version`, `scope`, `outcome`, `actor`, and `reason` are
+#: dropped: the receiver derives or ignores every one of them.
+_LEGACY_KEYS = {
+    "event_id": "id",
+    "action": "type",
+    "resource": "subject",
+    "observed_at": "occurred_at",
+    "details": "data",
+}
+
+
+def wire_envelope(envelope: dict[str, Any]) -> dict[str, Any]:
+    """The body to POST, upgrading a legacy stored envelope if needed."""
+    if all(envelope.get(field) for field in REQUIRED_FIELDS):
+        return {field: envelope[field] for field in REQUIRED_FIELDS}
+    upgraded = {key: envelope[key] for key in REQUIRED_FIELDS if key in envelope}
+    for legacy, current in _LEGACY_KEYS.items():
+        if not upgraded.get(current) and envelope.get(legacy):
+            upgraded[current] = envelope[legacy]
+    upgraded.setdefault("source", SOURCE)
+    data = upgraded.get("data")
+    # A legacy envelope predates the correlation_id requirement, but it
+    # carried the payload -- and the payload is where mutations recorded
+    # their correlation. Take it from there or not at all; audit-core is
+    # explicit that a synthesized correlation ties an event to an operation
+    # nobody observed.
+    if not upgraded.get("correlation_id") and isinstance(data, dict):
+        correlation = data.get("correlation_id")
+        if correlation:
+            upgraded["correlation_id"] = correlation
+    missing = [field for field in REQUIRED_FIELDS if not upgraded.get(field)]
+    if missing:
+        raise EnvelopeError(f"envelope missing required field(s): {', '.join(missing)}")
+    return upgraded
 
 
 @dataclass(frozen=True, slots=True)
 class DeliveryResult:
     event_id: str
-    status: str  # delivered | duplicate | retry | dead | skipped
+    status: str  # delivered | duplicate | retry | invalid | dead | skipped
     http_status: int | None = None
     detail: str = ""
 
@@ -79,8 +166,18 @@ class AuditCoreClient:
         )
 
     def post_event(self, envelope: dict[str, Any]) -> DeliveryResult:
-        event_id = str(envelope.get("event_id") or "")
-        headers: dict[str, str] = {"Content-Type": "application/json"}
+        event_id = str(envelope.get("id") or envelope.get("event_id") or "")
+        try:
+            body = wire_envelope(envelope)
+        except EnvelopeError as exc:
+            # Not sent. An unsendable envelope stays pending and visible
+            # rather than being recorded as a delivery outcome.
+            return DeliveryResult(event_id, "invalid", None, f"envelope:{exc}")
+        headers: dict[str, str] = {
+            "Content-Type": "application/json",
+            # The receiver rejects a body whose id disagrees with this header.
+            "Idempotency-Key": str(body["id"]),
+        }
         if self.token_file:
             try:
                 token = open(self.token_file, encoding="utf-8").read().strip()
@@ -90,17 +187,40 @@ class AuditCoreClient:
                 return DeliveryResult(event_id, "retry", None, "token_empty")
             headers["Authorization"] = f"Bearer {token}"
         try:
-            response = self._client.post("/v1/events", json=envelope, headers=headers)
+            response = self._client.post("/v1/events", json=body, headers=headers)
         except (httpx.HTTPError, OSError) as exc:
             return DeliveryResult(event_id, "retry", None, f"unreachable:{exc.__class__.__name__}")
         if response.status_code in (200, 202):
             status = "duplicate" if response.status_code == 200 else "delivered"
             return DeliveryResult(event_id, status, response.status_code)
-        if response.status_code in (400, 409):
-            return DeliveryResult(event_id, "dead", response.status_code, "rejected")
+        if response.status_code == 400:
+            # A 400 means the event is NOT in the archive: audit-core holds
+            # only an unchained dead letter. Treating that as terminal would
+            # mark our row handled and lose the event on both sides, so the
+            # row stays pending and the reason is recorded on it. A 400 is an
+            # integration defect to fix, not a delivery outcome.
+            return DeliveryResult(
+                event_id, "invalid", 400, f"rejected:{_reason(response)}"
+            )
+        if response.status_code == 409:
+            return DeliveryResult(event_id, "dead", 409, "rejected")
         if response.status_code in (401, 403):
             return DeliveryResult(event_id, "retry", response.status_code, "unauthorized")
         return DeliveryResult(event_id, "retry", response.status_code, "unavailable")
 
     def close(self) -> None:
         self._client.close()
+
+
+def _reason(response: httpx.Response) -> str:
+    """audit-core's rejection reason, so the outbox row says what to fix."""
+    try:
+        body = response.json()
+    except ValueError:
+        return "unparseable"
+    if isinstance(body, dict):
+        for key in ("reason", "error", "detail"):
+            value = body.get(key)
+            if isinstance(value, str) and value:
+                return value
+    return "unknown"

src/tenant_engine/postgres_store.py

@@ -19,7 +19,7 @@ except ImportError:  # pragma: no cover - exercised by the deployment guard
     ConnectionPool = None
     PoolTimeout = None
 
-from tenant_engine.audit_core import envelope_for, new_event_id
+from tenant_engine.audit_core import correlation_for, envelope_for, new_event_id
 from tenant_engine.domain import (
     CapabilityRole,
     PlanAssignment,
@@ -121,7 +121,13 @@ class PostgresTenantStore:
         with self._pool.connection() as conn:
             conn.execute("SELECT 1").fetchone()
 
-    def create_tenant(self, tenant: Tenant, *, authz: dict[str, Any] | None = None) -> None:
+    def create_tenant(
+        self,
+        tenant: Tenant,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
+    ) -> None:
         try:
             with self._pool.connection() as conn, conn.transaction():
                 conn.execute(
@@ -147,7 +153,12 @@ class PostgresTenantStore:
                     conn,
                     "tenant_created",
                     tenant.tenant_id,
-                    {"identifier": tenant.identifier, "grouping": tenant.grouping, **(authz or {})},
+                    {
+                        "identifier": tenant.identifier,
+                        "grouping": tenant.grouping,
+                        "correlation_id": correlation_id,
+                        **(authz or {}),
+                    },
                 )
         except StoreUnavailableError as exc:
             if isinstance(exc.__cause__, UniqueViolation):
@@ -254,7 +265,13 @@ class PostgresTenantStore:
             )
 
     def revoke_role(
-        self, *, tenant_id: str, grant_id: str, at: datetime, authz: dict[str, Any] | None = None
+        self,
+        *,
+        tenant_id: str,
+        grant_id: str,
+        at: datetime,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> RoleGrant:
         tenant = self.get_tenant(tenant_id)
         with self._pool.connection() as conn, conn.transaction():
@@ -270,7 +287,12 @@ class PostgresTenantStore:
                 conn,
                 "role_revoked",
                 tenant.tenant_id,
-                {"grant_id": grant_id, "role": grant.role.value, **(authz or {})},
+                {
+                    "grant_id": grant_id,
+                    "role": grant.role.value,
+                    "correlation_id": correlation_id or grant.correlation_id,
+                    **(authz or {}),
+                },
             )
             return grant
 
@@ -284,7 +306,11 @@ class PostgresTenantStore:
         return frozenset(CapabilityRole(row["role"]) for row in rows)
 
     def assign_plan(
-        self, assignment: PlanAssignment, *, authz: dict[str, Any] | None = None
+        self,
+        assignment: PlanAssignment,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> None:
         tenant = self.get_tenant(assignment.tenant_id)
         self._require_active(tenant, "assign a plan")
@@ -299,7 +325,11 @@ class PostgresTenantStore:
                 conn,
                 "plan_assigned",
                 tenant.tenant_id,
-                {"plan_id": assignment.plan_id, **(authz or {})},
+                {
+                    "plan_id": assignment.plan_id,
+                    "correlation_id": correlation_id,
+                    **(authz or {}),
+                },
             )
 
     def events_for(self, tenant_id: str) -> list[DomainEvent]:
@@ -548,7 +578,11 @@ class PostgresTenantStore:
     @staticmethod
     def _emit(conn: Any, event_type: str, tenant_id: str, payload: dict[str, Any]) -> None:
         event_id = str(payload.get("event_id") or new_event_id())
-        payload = {**payload, "event_id": event_id}
+        # audit-core requires a truthy correlation_id on every event
+        # (its docs/event-envelope.md). Normalize before the local
+        # insert so the local record and the emitted envelope carry
+        # the same id.
+        payload = {**payload, "event_id": event_id, "correlation_id": correlation_for(payload)}
         at = datetime.now(UTC)
         conn.execute(
             "INSERT INTO events (event_type, tenant_id, at, payload) VALUES (%s, %s, %s, %s)",
@@ -558,7 +592,8 @@ class PostgresTenantStore:
             event_id=event_id,
             event_type=event_type,
             tenant_id=tenant_id,
-            observed_at=at.isoformat(),
+            occurred_at=at.isoformat(),
+            correlation_id=payload["correlation_id"],
             payload=payload,
         )
         conn.execute(

src/tenant_engine/sqlite_store.py

@@ -8,7 +8,7 @@ from datetime import datetime
 from threading import RLock
 from typing import Any
 
-from tenant_engine.audit_core import envelope_for, new_event_id
+from tenant_engine.audit_core import correlation_for, envelope_for, new_event_id
 from tenant_engine.domain import (
     CapabilityRole,
     PlanAssignment,
@@ -126,7 +126,13 @@ class SQLiteTenantStore:
         with self._lock:
             self._db.execute("SELECT 1").fetchone()
 
-    def create_tenant(self, tenant: Tenant, *, authz: dict[str, Any] | None = None) -> None:
+    def create_tenant(
+        self,
+        tenant: Tenant,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
+    ) -> None:
         with self._lock, self._db:
             try:
                 self._db.execute(
@@ -156,6 +162,7 @@ class SQLiteTenantStore:
                 {
                     "identifier": tenant.identifier,
                     "grouping": tenant.grouping,
+                    "correlation_id": correlation_id,
                     **(authz or {}),
                 },
             )
@@ -453,7 +460,13 @@ class SQLiteTenantStore:
             )
 
     def revoke_role(
-        self, *, tenant_id: str, grant_id: str, at: datetime, authz: dict[str, Any] | None = None
+        self,
+        *,
+        tenant_id: str,
+        grant_id: str,
+        at: datetime,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> RoleGrant:
         tenant = self.get_tenant(tenant_id)
         with self._lock:
@@ -471,7 +484,12 @@ class SQLiteTenantStore:
             self._emit(
                 "role_revoked",
                 tenant.tenant_id,
-                {"grant_id": grant_id, "role": grant.role.value, **(authz or {})},
+                {
+                    "grant_id": grant_id,
+                    "role": grant.role.value,
+                    "correlation_id": correlation_id or grant.correlation_id,
+                    **(authz or {}),
+                },
             )
         return grant
 
@@ -485,7 +503,11 @@ class SQLiteTenantStore:
         return frozenset(CapabilityRole(row["role"]) for row in rows)
 
     def assign_plan(
-        self, assignment: PlanAssignment, *, authz: dict[str, Any] | None = None
+        self,
+        assignment: PlanAssignment,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> None:
         tenant = self.get_tenant(assignment.tenant_id)
         self._require_active(tenant, "assign a plan")
@@ -497,7 +519,11 @@ class SQLiteTenantStore:
             self._emit(
                 "plan_assigned",
                 tenant.tenant_id,
-                {"plan_id": assignment.plan_id, **(authz or {})},
+                {
+                    "plan_id": assignment.plan_id,
+                    "correlation_id": correlation_id,
+                    **(authz or {}),
+                },
             )
 
     def events_for(self, tenant_id: str) -> list[DomainEvent]:
@@ -590,7 +616,11 @@ class SQLiteTenantStore:
 
     def _emit(self, event_type: str, tenant_id: str, payload: dict) -> None:
         event_id = str(payload.get("event_id") or new_event_id())
-        payload = {**payload, "event_id": event_id}
+        # audit-core requires a truthy correlation_id on every event
+        # (its docs/event-envelope.md). Normalize before the local
+        # insert so the local record and the emitted envelope carry
+        # the same id.
+        payload = {**payload, "event_id": event_id, "correlation_id": correlation_for(payload)}
         now = datetime.now().astimezone()
         self._db.execute(
             "INSERT INTO events(event_type,tenant_id,at,payload) VALUES(?,?,?,?)",
@@ -600,7 +630,8 @@ class SQLiteTenantStore:
             event_id=event_id,
             event_type=event_type,
             tenant_id=tenant_id,
-            observed_at=now.isoformat(),
+            occurred_at=now.isoformat(),
+            correlation_id=payload["correlation_id"],
             payload=payload,
         )
         self._db.execute(

src/tenant_engine/store.py

@@ -5,7 +5,7 @@ from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from typing import Any, Protocol
 
-from tenant_engine.audit_core import envelope_for, new_event_id
+from tenant_engine.audit_core import correlation_for, envelope_for, new_event_id
 from tenant_engine.domain import (
     CapabilityRole,
     PlanAssignment,
@@ -109,20 +109,36 @@ class TenantStore(Protocol):
     tenant_roles claim, KEY-WP-0005-T02).
     """
 
-    def create_tenant(self, tenant: Tenant, *, authz: dict[str, Any] | None = None) -> None: ...
+    def create_tenant(
+        self,
+        tenant: Tenant,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
+    ) -> None: ...
 
     def get_tenant(self, tenant_id: str) -> Tenant: ...
 
     def grant_role(self, grant: RoleGrant, *, authz: dict[str, Any] | None = None) -> None: ...
 
     def revoke_role(
-        self, *, tenant_id: str, grant_id: str, at: datetime, authz: dict[str, Any] | None = None
+        self,
+        *,
+        tenant_id: str,
+        grant_id: str,
+        at: datetime,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> RoleGrant: ...
 
     def active_roles(self, tenant_id: str) -> frozenset[CapabilityRole]: ...
 
     def assign_plan(
-        self, assignment: PlanAssignment, *, authz: dict[str, Any] | None = None
+        self,
+        assignment: PlanAssignment,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> None: ...
 
     def events_for(self, tenant_id: str) -> list[DomainEvent]: ...
@@ -253,7 +269,13 @@ class InMemoryTenantStore:
         self._overrides: dict[str, dict[str, LimitValue]] = {}
         self._guardrail_changes: list[GuardrailChange] = []
 
-    def create_tenant(self, tenant: Tenant, *, authz: dict[str, Any] | None = None) -> None:
+    def create_tenant(
+        self,
+        tenant: Tenant,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
+    ) -> None:
         if tenant.tenant_id in self._tenants:
             raise TenantAlreadyExistsError(tenant.tenant_id)
         if tenant.identifier in self._by_identifier:
@@ -264,7 +286,12 @@ class InMemoryTenantStore:
         self._emit(
             "tenant_created",
             tenant.tenant_id,
-            {"identifier": tenant.identifier, "grouping": tenant.grouping, **(authz or {})},
+            {
+                "identifier": tenant.identifier,
+                "grouping": tenant.grouping,
+                "correlation_id": correlation_id,
+                **(authz or {}),
+            },
         )
 
     def get_tenant(self, tenant_id: str) -> Tenant:
@@ -293,6 +320,7 @@ class InMemoryTenantStore:
         grant_id: str,
         at: datetime,
         authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> RoleGrant:
         resolved = self._resolve(tenant_id)
         try:
@@ -304,7 +332,12 @@ class InMemoryTenantStore:
         self._emit(
             "role_revoked",
             resolved,
-            {"grant_id": grant_id, "role": revoked.role.value, **(authz or {})},
+            {
+                "grant_id": grant_id,
+                "role": revoked.role.value,
+                "correlation_id": correlation_id or grant.correlation_id,
+                **(authz or {}),
+            },
         )
         return revoked
 
@@ -315,7 +348,11 @@ class InMemoryTenantStore:
         )
 
     def assign_plan(
-        self, assignment: PlanAssignment, *, authz: dict[str, Any] | None = None
+        self,
+        assignment: PlanAssignment,
+        *,
+        authz: dict[str, Any] | None = None,
+        correlation_id: str | None = None,
     ) -> None:
         resolved = self._resolve(assignment.tenant_id)
         self._require_active(resolved, "assign a plan")
@@ -323,7 +360,11 @@ class InMemoryTenantStore:
         self._emit(
             "plan_assigned",
             resolved,
-            {"plan_id": assignment.plan_id, **(authz or {})},
+            {
+                "plan_id": assignment.plan_id,
+                "correlation_id": correlation_id,
+                **(authz or {}),
+            },
         )
 
     def events_for(self, tenant_id: str) -> list[DomainEvent]:
@@ -494,7 +535,11 @@ class InMemoryTenantStore:
 
     def _emit(self, event_type: str, tenant_id: str, payload: dict[str, Any]) -> None:
         event_id = str(payload.get("event_id") or new_event_id())
-        payload = {**payload, "event_id": event_id}
+        # audit-core requires a truthy correlation_id on every event
+        # (its docs/event-envelope.md). Normalize before the local
+        # insert so the local record and the emitted envelope carry
+        # the same id.
+        payload = {**payload, "event_id": event_id, "correlation_id": correlation_for(payload)}
         at = datetime.now(UTC)
         self._events.append(
             DomainEvent(event_type=event_type, tenant_id=tenant_id, at=at, payload=payload)
@@ -503,7 +548,8 @@ class InMemoryTenantStore:
             event_id=event_id,
             event_type=event_type,
             tenant_id=tenant_id,
-            observed_at=at.isoformat(),
+            occurred_at=at.isoformat(),
+            correlation_id=payload["correlation_id"],
             payload=payload,
         )
         self._outbox[event_id] = OutboxRow(