Phase 0: per-call GatewayCallAudit on the canonical inference gateway (#45)

* feat(gateway): per-call GatewayCallAudit on the canonical inference chokepoint

Memory mesh is the single inference gateway both estate planes route through,
so caller identity, model, token usage/cost, and epistemic labeling should be
recorded here once rather than per-consumer. This adds that audit record.

- schemas/gateway-call-audit.schema.json: GatewayCallAudit v0.1 (Draft 2020-12)
- adapters/litellm/gateway_audit.py: pure builder; raw completions default to
  epistemicLevel=speculative (never defaulted up), hash-bound receipt
- adapters/litellm/memory_mesh_hooks.py: emit audit on success AND on failure
  (async_log_failure_event was a no-op — failures went unaudited); best-effort
  POST to memoryd /v1/audit, non-fatal; x-memory-mesh-audit-id response header
- scripts/validate_gateway_call_audit.py + examples: house-style validator
- Makefile: validate-gateway-call-audit wired into validate
- LICENSE: add MIT (repo had none)

Part of Integration Mission epic SocioProphet/socioprophet#490 (Phase 0).
validate-gateway-call-audit green; 9 builder tests pass. Closes #44

* fix(gateway): audit every successful completion, not just recall-hit calls

Adversarial review found the audit call sat behind three early returns
(writeback disabled, no recall envelope, empty text), so a successful call
with no recall hits emitted no audit — defeating 'every call is audited'.

Restructure: writeback stays independently gated, but the audit always fires
on success (rebuilding a minimal envelope when recall produced no hits). Add
hook-level tests (litellm stubbed) proving audit fires for the no-hits and
writeback-disabled cases plus the failure event.

* fix(gateway): builder robustness from Copilot review

- _normalize_usage: a non-int-castable total_tokens fell outside the try and
  would raise into the gateway call path; coerce defensively, fall back to
  prompt+completion.
- _caller_from_envelope: a missing/blank user_id produced an empty string,
  violating the schema (caller.user_id minLength 1); use an explicit 'unknown'
  sentinel so output stays schema-valid.
- build_gateway_call_audit: validate call_type against the schema enum
  (fail fast) to keep the 'schema-valid output' contract.
Add tests for all three.

Author: mdheller

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 SocioProphet
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

@@ -1,6 +1,6 @@
 PYTHON ?= python
 
-.PHONY: validate-upstreams validate-python validate-deploy-assets validate-agent-learning-proposal validate-scenario-learning-binding validate-governed-learning-lifecycle validate-workspace-recall-promotion validate-channel-provenance-write-gate validate-wallguard-memory-compartment-gate validate-memory-distribution-grant validate-prophet-mesh-scope-mirror validate local-preflight local-up local-smoke local-debug local-down
+.PHONY: validate-upstreams validate-python validate-deploy-assets validate-agent-learning-proposal validate-scenario-learning-binding validate-governed-learning-lifecycle validate-workspace-recall-promotion validate-channel-provenance-write-gate validate-wallguard-memory-compartment-gate validate-memory-distribution-grant validate-gateway-call-audit validate-prophet-mesh-scope-mirror validate local-preflight local-up local-smoke local-debug local-down
 
 validate-upstreams:
 	$(PYTHON) scripts/validate_upstreams.py third_party/upstreams.lock.yaml
@@ -36,7 +36,11 @@ validate-wallguard-memory-compartment-gate:
 validate-memory-distribution-grant:
 	$(PYTHON) scripts/validate_memory_distribution_grant.py
 
-validate: validate-upstreams validate-python validate-deploy-assets validate-wallguard-memory-compartment-gate validate-memory-distribution-grant validate-prophet-mesh-scope-mirror
+validate-gateway-call-audit:
+	$(PYTHON) scripts/validate_gateway_call_audit.py
+	$(PYTHON) -m pytest -q adapters/litellm/tests/test_gateway_audit.py
+
+validate: validate-upstreams validate-python validate-deploy-assets validate-wallguard-memory-compartment-gate validate-memory-distribution-grant validate-gateway-call-audit validate-prophet-mesh-scope-mirror
 
 local-preflight:
 	bash deploy/local/scripts/preflight-podman-m2.sh

adapters/litellm/gateway_audit.py

@@ -0,0 +1,152 @@
+"""Gateway call audit builder.
+
+Memory mesh is the canonical inference chokepoint. Every completion that passes
+through the LiteLLM gateway should produce exactly one `GatewayCallAudit` record
+carrying caller identity, model, token usage / cost, and an epistemic label —
+so labeling, cost signal, and audit happen here once rather than per-consumer.
+
+This module is a pure builder (no I/O) so it is trivially testable and can be
+called from both the success and failure paths of the gateway hook.
+
+Epistemic default: a raw gateway completion is `speculative` — an ungrounded
+model output — unless the caller explicitly supplies a higher-warrant level
+(e.g. a promoted/proved model route). We never default *up*.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Mapping, Optional
+
+SCHEMA_VERSION = "memory-mesh.gateway-call-audit.v0.1"
+RECORD_TYPE = "GatewayCallAudit"
+
+ALLOWED_EPISTEMIC_LEVELS = {
+    "proved",
+    "bounded",
+    "empirical",
+    "synthetic",
+    "speculative",
+    "rejected",
+}
+DEFAULT_EPISTEMIC_LEVEL = "speculative"
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _caller_from_envelope(envelope: Mapping[str, Any]) -> dict[str, str]:
+    # The schema requires caller.user_id to be non-empty (minLength: 1); a
+    # missing/blank id becomes an explicit "unknown" sentinel so the record
+    # stays schema-valid and the gap is visible rather than silently empty.
+    user_id = str(envelope.get("user_id") or "").strip() or "unknown"
+    return {
+        "user_id": user_id,
+        "agent_id": str(envelope.get("agent_id") or ""),
+        "run_id": str(envelope.get("run_id") or ""),
+        "workload_id": str(envelope.get("workload_id") or ""),
+        "source_interface": str(envelope.get("source_interface") or ""),
+    }
+
+
+def _normalize_usage(usage: Optional[Mapping[str, Any]]) -> Optional[dict[str, int]]:
+    if not usage:
+        return None
+    try:
+        prompt = int(usage.get("prompt_tokens", 0) or 0)
+        completion = int(usage.get("completion_tokens", 0) or 0)
+    except (TypeError, ValueError):
+        return None
+    total_raw = usage.get("total_tokens")
+    if total_raw is None:
+        total = prompt + completion
+    else:
+        # Defensive: a non-int-castable total must not raise into the gateway
+        # call path — fall back to prompt+completion instead of breaking the call.
+        try:
+            total = int(total_raw)
+        except (TypeError, ValueError):
+            total = prompt + completion
+    return {
+        "prompt_tokens": prompt,
+        "completion_tokens": completion,
+        "total_tokens": total,
+    }
+
+
+def _resolve_epistemic_level(requested: Optional[str]) -> str:
+    if requested and requested in ALLOWED_EPISTEMIC_LEVELS:
+        return requested
+    return DEFAULT_EPISTEMIC_LEVEL
+
+
+def _receipt_hash(core: Mapping[str, Any]) -> str:
+    canonical = json.dumps(core, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return "sha256:" + hashlib.sha256(canonical).hexdigest()
+
+
+def build_gateway_call_audit(
+    *,
+    envelope: Mapping[str, Any],
+    call_type: str,
+    outcome: str,
+    model: Optional[str] = None,
+    usage: Optional[Mapping[str, Any]] = None,
+    cost_usd: Optional[float] = None,
+    epistemic_level: Optional[str] = None,
+    recalled_count: int = 0,
+    written: int = 0,
+    error: Optional[str] = None,
+    occurred_at: Optional[str] = None,
+    call_id: Optional[str] = None,
+) -> dict[str, Any]:
+    """Build a schema-valid GatewayCallAudit record.
+
+    `outcome` is "success" or "failure"; a failure record must carry an error
+    string. The returned record is stable and hash-bound via `receipt_hash`.
+    """
+    if outcome not in {"success", "failure"}:
+        raise ValueError(f"outcome must be 'success' or 'failure', got {outcome!r}")
+    if call_type not in {"completion", "text_completion"}:
+        # Keep the "schema-valid output" contract: fail fast if a new call path
+        # passes a type the schema does not allow, rather than emitting an
+        # invalid record.
+        raise ValueError(
+            f"call_type must be 'completion' or 'text_completion', got {call_type!r}"
+        )
+    if outcome == "failure" and not error:
+        # Fail closed on our own audit: a failure with no reason is not auditable.
+        error = "unspecified_gateway_failure"
+
+    caller = _caller_from_envelope(envelope)
+    normalized_usage = _normalize_usage(usage)
+    level = _resolve_epistemic_level(epistemic_level)
+    when = occurred_at or _utc_now_iso()
+    cid = call_id or f"gateway-call:{uuid.uuid4()}"
+
+    core = {
+        "callId": cid,
+        "call_type": call_type,
+        "outcome": outcome,
+        "caller": caller,
+        "model": model,
+        "usage": normalized_usage,
+        "cost_usd": cost_usd,
+        "epistemicLevel": level,
+        "recalled_count": int(recalled_count),
+        "written": int(written),
+        "error": error,
+        "occurred_at": when,
+    }
+
+    record = {
+        "schemaVersion": SCHEMA_VERSION,
+        "recordType": RECORD_TYPE,
+        **core,
+        "receipt_hash": _receipt_hash(core),
+    }
+    return record

adapters/litellm/memory_mesh_hooks.py

@@ -1,12 +1,22 @@
 from __future__ import annotations
 
 import os
+import sys
+from pathlib import Path
 from typing import Any, Dict, Iterable, Literal, Optional
 
 import httpx
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.proxy.proxy_server import DualCache, UserAPIKeyAuth
 
+# gateway_audit is a sibling module; ensure it is importable regardless of how
+# LiteLLM loads this hook (by file path or by module name).
+_ADAPTER_DIR = str(Path(__file__).resolve().parent)
+if _ADAPTER_DIR not in sys.path:
+    sys.path.insert(0, _ADAPTER_DIR)
+
+from gateway_audit import build_gateway_call_audit  # noqa: E402
+
 
 class MemoryMeshHook(CustomLogger):
     def __init__(self) -> None:
@@ -89,46 +99,126 @@ async def async_pre_call_hook(
         return data
 
     async def async_post_call_success_hook(self, data: dict, user_api_key_dict: UserAPIKeyAuth, response: Any) -> Any:
-        if not self.writeback_enabled:
-            return response
-
-        metadata = data.get("metadata") or {}
+        metadata = data.setdefault("metadata", {})
         envelope = metadata.get("memory_mesh_envelope")
         if not envelope:
-            return self._attach_headers(response, recalled=0, written=0)
+            # Recall may not have produced hits (or was skipped for a non-completion),
+            # so no envelope was stashed. Rebuild a minimal one so that EVERY
+            # successful completion is audited, not just recall-hit calls.
+            user_id = data.get("user")
+            envelope = self._build_envelope(data=data, user_id=str(user_id)) if user_id else None
 
         recalled = int(metadata.get("memory_mesh_hit_count") or 0)
-        assistant_text = self._extract_assistant_text(response)
-        user_text = self._build_recall_query(data)
-        if not assistant_text and not user_text:
-            return self._attach_headers(response, recalled=recalled, written=0)
-
-        content = self._build_interaction_record(user_text=user_text, assistant_text=assistant_text)
-        payload = {
-            "envelope": envelope,
-            "content": content,
-            "memory_class": self.default_writeback_class,
-            "persist_to_backend": True,
-            "metadata": {
-                "model": data.get("model"),
-                "recalled_count": recalled,
-                "source": "litellm-hook",
-            },
-            "tags": ["litellm", "interaction"],
-        }
 
+        # Writeback is independently gated (flag + envelope + non-empty text);
+        # it must never gate the audit below.
         written = 0
+        if self.writeback_enabled and envelope:
+            assistant_text = self._extract_assistant_text(response)
+            user_text = self._build_recall_query(data)
+            if assistant_text or user_text:
+                content = self._build_interaction_record(user_text=user_text, assistant_text=assistant_text)
+                payload = {
+                    "envelope": envelope,
+                    "content": content,
+                    "memory_class": self.default_writeback_class,
+                    "persist_to_backend": True,
+                    "metadata": {
+                        "model": data.get("model"),
+                        "recalled_count": recalled,
+                        "source": "litellm-hook",
+                    },
+                    "tags": ["litellm", "interaction"],
+                }
+                try:
+                    async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+                        write_response = await client.post(f"{self.base_url}/v1/write", json=payload, headers=self.headers)
+                        write_response.raise_for_status()
+                        written = 1
+                except Exception as exc:  # pragma: no cover
+                    metadata["memory_mesh_write_error"] = str(exc)
+
+        # Audit ALWAYS fires on a successful completion — independent of recall
+        # hits or writeback. Only a request with no resolvable caller is skipped.
+        if envelope is None:
+            return self._attach_headers(response, recalled=recalled, written=written)
+
+        audit = build_gateway_call_audit(
+            envelope=envelope,
+            call_type="completion",
+            outcome="success",
+            model=data.get("model"),
+            usage=self._extract_usage(response),
+            cost_usd=self._extract_cost(response),
+            epistemic_level=metadata.get("epistemic_level"),
+            recalled_count=recalled,
+            written=written,
+        )
+        await self._emit_audit(audit)
+        metadata["memory_mesh_audit_id"] = audit["callId"]
+        return self._attach_headers(response, recalled=recalled, written=written, audit_id=audit["callId"])
+
+    async def async_log_failure_event(self, kwargs: dict, response_obj: Any, start_time: Any, end_time: Any) -> None:
+        # A failed gateway call must still be audited (fail-closed on observability).
+        data = kwargs.get("litellm_params", {}).get("metadata", {}) or {}
+        envelope = data.get("memory_mesh_envelope")
+        if not isinstance(envelope, dict):
+            model = kwargs.get("model")
+            envelope = {
+                "user_id": str(kwargs.get("user") or "unknown"),
+                "agent_id": self.default_agent_id,
+                "run_id": f"litellm:{kwargs.get('user') or 'unknown'}",
+                "workload_id": self.default_workload_id,
+                "source_interface": self.default_source_interface,
+            }
+        else:
+            model = (kwargs.get("model") or envelope.get("metadata", {}).get("request_model"))
+        exception = kwargs.get("exception")
+        audit = build_gateway_call_audit(
+            envelope=envelope,
+            call_type="completion",
+            outcome="failure",
+            model=model,
+            epistemic_level=data.get("epistemic_level"),
+            error=str(exception) if exception else "gateway_call_failed",
+        )
+        await self._emit_audit(audit)
+        return None
+
+    async def _emit_audit(self, audit: Dict[str, Any]) -> None:
+        """Best-effort POST of the audit record to memoryd; never fatal to the call."""
         try:
             async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
-                write_response = await client.post(f"{self.base_url}/v1/write", json=payload, headers=self.headers)
-                write_response.raise_for_status()
-                written = 1
-        except Exception as exc:  # pragma: no cover
-            metadata["memory_mesh_write_error"] = str(exc)
+                response = await client.post(f"{self.base_url}/v1/audit", json=audit, headers=self.headers)
+                response.raise_for_status()
+        except Exception:  # pragma: no cover - audit transport is best-effort
+            pass
 
-        return self._attach_headers(response, recalled=recalled, written=written)
+    @staticmethod
+    def _extract_usage(response: Any) -> Optional[Dict[str, Any]]:
+        usage = getattr(response, "usage", None)
+        if usage is None and isinstance(response, dict):
+            usage = response.get("usage")
+        if usage is None:
+            return None
+        if not isinstance(usage, dict):
+            usage = {
+                "prompt_tokens": getattr(usage, "prompt_tokens", None),
+                "completion_tokens": getattr(usage, "completion_tokens", None),
+                "total_tokens": getattr(usage, "total_tokens", None),
+            }
+        return {k: v for k, v in usage.items() if v is not None} or None
 
-    async def async_log_failure_event(self, kwargs: dict, response_obj: Any, start_time: Any, end_time: Any) -> None:
+    @staticmethod
+    def _extract_cost(response: Any) -> Optional[float]:
+        hidden = getattr(response, "_hidden_params", None)
+        if isinstance(hidden, dict):
+            cost = hidden.get("response_cost")
+            if cost is not None:
+                try:
+                    return float(cost)
+                except (TypeError, ValueError):
+                    return None
         return None
 
     @staticmethod
@@ -209,10 +299,12 @@ def _build_interaction_record(user_text: str, assistant_text: str) -> str:
         return "\n".join(parts)
 
     @staticmethod
-    def _attach_headers(response: Any, recalled: int, written: int) -> Any:
+    def _attach_headers(response: Any, recalled: int, written: int, audit_id: Optional[str] = None) -> Any:
         additional_headers = getattr(response, "_hidden_params", {}).get("additional_headers", {}) or {}
         additional_headers["x-memory-mesh-recalled"] = str(recalled)
         additional_headers["x-memory-mesh-written"] = str(written)
+        if audit_id:
+            additional_headers["x-memory-mesh-audit-id"] = audit_id
         if not hasattr(response, "_hidden_params"):
             response._hidden_params = {}
         response._hidden_params["additional_headers"] = additional_headers