Add unit tests for audit logging and sinks, enhance config validation

CHANGELOG.md

@@ -29,6 +29,32 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Added: audit log
+
+- **`audit` key in `10xgraph.json`** (`enabled`, `sinks`, `include_reads`; off by default, unknown
+  keys refuse to boot). Records who accessed or changed what, when, and whether it was allowed, as
+  metadata only: ids and results, never prompts, answers, message text or file contents. Event
+  types: `access` (`allowed`, `denied`, `not_found`), `auth` (`failed` with a reason) and `change`
+  (`success`, with a `count` where known). `include_reads: false` drops allowed access events;
+  denials, auth failures and changes are always kept.
+- **Sinks.** `"log"` writes one JSON line per event on stdout (logger `tenxgraph_api.audit`),
+  whatever `LOG_FORMAT`. `"publisher"` also sends each event through the app's events publisher
+  (Kafka, Redis, RabbitMQ, custom); it is best effort and events still queued at SIGTERM may be
+  lost, so the log sink is the record. Observability publishers (OTEL, Logfire, Phoenix) are
+  skipped, and a `"publisher"` entry with no events publisher logs a warning at boot and is
+  skipped. `"module:Class"` loads your own sink once at boot (a `BaseAuditSink` subclass, an
+  instance or a zero-argument factory); anything else refuses to boot.
+- **`BaseAuditSink` and `AuditEvent`** are exported from `tenxgraph_api`. A sink that raises is
+  logged at ERROR and the request continues; with metrics on it also counts into
+  `tenxgraph_audit_sink_errors_total` (label `sink`, the sink class name).
+- **`LOG_FORMAT=json`** writes every log line as one JSON object, with request id, transport and
+  user id when known; secrets are still masked.
+- **`10xgraph config`** has an Audit log section.
+- **`10xgraph init` production template** turns the audit log on, and **`10xgraph build`** sets
+  `LOG_FORMAT=json` in the Dockerfile, Kubernetes manifest and Docker Compose file.
+- **`10xgraph audit`** has an Audit log check that warns when `MODE=production` and
+  `audit.enabled` is not true.
+
 ### Breaking changes: auth, RBAC and reach
 
 Needs the `10xgraph` release with `Identity` and `Reach` (`tenxgraph.core.authz`). The unreleased

CLAUDE.md

@@ -45,7 +45,7 @@ Importable package: `tenxgraph_api/`. Two halves:
 | `tenxgraph_api/src/app/` | The FastAPI app. `main.py` + `loader.py` (build app from `10xgraph.json`), `routers/` (graph, checkpointer, store, media, ping, ag_ui), `core/auth/`, `core/config/`, `core/middleware/` (rate_limit, security_headers, request_limits), `tasks/`, `utils/`, `worker.py` |
 
 Public exports from the package root (`from tenxgraph_api import ...`): `BaseAuth`,
-`SnowFlakeIdGenerator`, `ThreadNameGenerator`.
+`SnowFlakeIdGenerator`, `ThreadNameGenerator`, `BaseAuditSink`, `AuditEvent`.
 
 ## CLI commands (verified against `cli/main.py`)
 
@@ -99,6 +99,7 @@ Parsed by `tenxgraph_api/src/app/core/config/graph_config.py`. Supported keys:
 | `websocket` | `enabled` (`false` leaves out `/v1/graph/ws` and `/v1/graph/live`; `true` without `websockets`/`wsproto` refuses to boot; unset keeps them and warns if no library; `routers/setup_router.py::_graph_router`), `max_connections` (1000), `max_connections_per_user` (10), `realtime_models` (models a `/v1/graph/live` client may pick), `realtime_client_overrides` (init options a client may set; default `voice`, `modalities`, `vad`; add `system_prompt` only for trusted callers) |
 | `runs` | `RunsConfig`: `thread_ids` (`server` default = only existing ids may be named on REST invoke/stream, `/v1/graph/ws` and checkpointer writes, a new thread is started by omitting the id; enforced by `require_thread` with auth on or off. AG-UI and `/v1/graph/live` follow the same rule (a blank or `new` id gets a server id). `client` = any request may name a new thread, explicit opt-in), `config_keys` / `state_keys` (allow-lists for request config and `initial_state`/PUT state; 422 on other keys; `thread_id`, `run_id` always allowed), `max_concurrent_per_user` (HTTP invoke/stream/AG-UI runs in flight, default 10, 429 over it; `routers/graph/run_limits.py`), `timeout` (seconds; invoke 504, streams end with an error chunk), `max_parallel_tool_calls` (positive int, default unset = core's default 10 or env `TENXGRAPH_MAX_PARALLEL_TOOL_CALLS`; set into every run's config in `_prepare_input`, reserved from clients) |
 | `cache_store.token_revocation` | `true`, `false`, `null` or `{"enabled", "token_lifetime", "user_ttl"}` (`graph_config.TokenRevocationConfig`; a string is a config error). Read through `GraphConfig.token_revocation`. Revocations live in the cache store (`core/auth/revocation.py`: `revoke_token`, `revoke_user`, `is_revoked`), checked by `RequirePermission` (401), before every `/v1/graph/ws` run, and before and every 30 s during a `/v1/graph/live` session (`REALTIME_RECHECK_INTERVAL`, close 1008); cache store errors fail closed (503), and an unreachable cache store refuses the boot |
+| `audit` | `core/audit/config.py::AuditConfig` (unknown keys refuse to boot): `enabled` (false), `sinks` (`["log"]`: JSON lines on stdout via logger `tenxgraph_api.audit`, own handler, never propagates; `"publisher"` sends each event through the app's events publisher, best effort, skipping AI tracing publishers (OTEL, Logfire, Phoenix), and is skipped with a boot warning when no events publisher is bound (`core/audit/sinks.py`); `"module:Class"` loads a `BaseAuditSink` instance, subclass or zero-arg factory once at boot), `include_reads` (true; false drops allowed access events). One emit function `core/audit::audit_event`, configured once in the lifespan (`configure_audit`). Hooks: `permissions.py` (`_authenticate`/`_identity` auth failures, `_authorize` access allowed/denied for `RequirePermission` and `authorize_resource`), `core/threads.py::require_thread` and the `ThreadNotFoundError` 404 handler (not_found, deduped by marking the exception), `routers/graph/router.py::_ws_credential_error` (WS re-checks); `change`/`success` events after successful writes and deletes in the checkpointer, store, media, graph (stop, fix) and tasks routers, and in `core/auth/revocation.py` (`tokens:revoke`, `users:revoke`, also from `10xgraph revoke`). Request id, transport and client ip come from `core/request_context.py`, bound by the request-id middleware |
 | `media` | `delete_with_thread` (default true): delete an uploaded file when the last message or thread that references it is deleted. Links are recorded in the file's `threads` metadata (`MediaService.link_threads` / `unlink_threads`); files without a `threads` key (uploaded before link tracking) and stores without `update_metadata` (core older than the release that added it) are never auto-deleted. `false` still records links. Parsed by `graph_config.MediaConfig` |
 
 `cache_store.rate_limit` object (read through `GraphConfig.rate_limit`, None when absent or

README.md

@@ -7,6 +7,7 @@
 [![PyPI](https://img.shields.io/pypi/v/10xgraph-api?color=blue)](https://pypi.org/project/10xgraph-api/)
 [![Python](https://img.shields.io/pypi/pyversions/10xgraph-api)](https://pypi.org/project/10xgraph-api/)
 [![License](https://img.shields.io/github/license/10xGraph/10xgraph-api)](https://github.com/10xGraph/10xgraph-api/blob/main/LICENSE)
+[![Verified on FreeAPI](https://findfreeapi.com/api/badge/10xgraph-api)](https://findfreeapi.com/en/service/10xgraph-api)
 
 `10xgraph-api` generates the production server around a [10xGraph](https://github.com/10xGraph/10xGraph) agent. Point `10xgraph.json` at your compiled graph and it serves REST, NDJSON streaming, WebSocket and realtime-audio endpoints with JWT or custom auth, scoped authorization on every endpoint, owner-only threads and rate limiting, then writes your Docker Compose and Kubernetes files. MIT licensed and self-hosted.
 

config-editor-ui/src/audit.jsx

@@ -0,0 +1,82 @@
+import { commit, isObject, issueAt, store } from "./store.js";
+import { Chips, FieldIssue, Toggle } from "./ui.jsx";
+
+const note = "mt-2 max-w-prose text-[13px] leading-5 text-muted";
+const audit = () => (isObject(store.config.audit) ? store.config.audit : {});
+const isOn = () => audit().enabled === true;
+const sinks = () => (Array.isArray(audit().sinks) ? audit().sinks : ["log"]);
+const custom = () => sinks().filter((s) => s !== "log" && s !== "publisher");
+
+const edit = (fn) => commit((s) => {
+  const a = isObject(s.config.audit) ? s.config.audit : {};
+  fn(a);
+  s.config.audit = a;
+});
+
+function setOn(on) {
+  if (!on) { commit((s) => { delete s.config.audit; }); return; }
+  edit((a) => {
+    a.enabled = true;
+    a.sinks = ["log"];
+    a.include_reads = true;
+  });
+}
+
+// "log" is always kept, so the file never ends up with an audit log that goes nowhere.
+function setSinks({ publisher = sinks().includes("publisher"), extra = custom() }) {
+  edit((a) => { a.sinks = ["log", ...(publisher ? ["publisher"] : []), ...extra]; });
+}
+
+export function AuditWidget() {
+  const on = isOn();
+  const reads = audit().include_reads !== false;
+  const publisher = sinks().includes("publisher");
+  return (
+    <div>
+      <div class="flex items-center gap-3">
+        <Toggle id="f-audit-enabled" on={on} label="Record an audit log" onChange={setOn} />
+        <label for="f-audit-enabled" class="cursor-pointer text-sm font-semibold text-ink">Record an audit log</label>
+      </div>
+      <p class={note}>{on ? "Events are written as JSON lines on stdout." : "No audit events are recorded."}</p>
+      <FieldIssue issue={issueAt("audit.enabled")} />
+      {on ? (
+        <div class="mt-5 rounded-xl border border-line bg-surface px-5 pt-1 pb-5">
+          <div class="pt-4">
+            <h3 class="text-sm font-semibold text-ink">Sinks</h3>
+            <p class={note}>
+              <code class="font-mono text-xs text-body">log</code> is always on: one JSON line per event on
+              stdout, for your log pipeline to collect.
+            </p>
+            <div class="mt-3 flex items-center gap-3">
+              <Toggle id="f-audit-publisher" on={publisher} label="Publisher" onChange={(v) => setSinks({ publisher: v })} />
+              <label for="f-audit-publisher" class="cursor-pointer text-sm font-semibold text-ink">Publisher</label>
+            </div>
+            <p class={note}>Also send events through the configured publisher. Best effort.</p>
+            <label for="f-audit-sinks" class="mt-4 block text-sm font-semibold text-ink">Custom sinks</label>
+            <div class="mt-2 max-w-[520px]">
+              <Chips
+                id="f-audit-sinks" values={custom()} placeholder="my_pkg.audit:MySink"
+                onChange={(next) => setSinks({ extra: next })}
+              />
+            </div>
+            <p class={note}>A <code class="font-mono text-xs text-body">module:Class</code> that extends BaseAuditSink.</p>
+            <FieldIssue issue={issueAt("audit.sinks")} />
+          </div>
+          <div class="mt-5 border-t border-line pt-4">
+            <div class="flex items-center gap-3">
+              <Toggle id="f-audit-include_reads" on={reads} label="Include allowed reads" onChange={(v) => edit((a) => { a.include_reads = v; })} />
+              <label for="f-audit-include_reads" class="cursor-pointer text-sm font-semibold text-ink">Include allowed reads</label>
+            </div>
+            <p class={note}>Off keeps denials, auth failures and changes, and drops allowed reads.</p>
+            <FieldIssue issue={issueAt("audit.include_reads")} />
+          </div>
+        </div>
+      ) : null}
+    </div>
+  );
+}
+
+export function auditSummary(offText) {
+  if (!isOn()) return { text: offText("audit") };
+  return { text: audit().include_reads === false ? "On, denials and changes only." : "On, all access and changes." };
+}

config-editor-ui/src/icons.js

@@ -41,7 +41,7 @@ export const ICONS = {
 export const SECTION_ICONS = {
   "get-running": "flag", environment: "terminal",
   core: "flag", auth: "lock", cache_store: "gauge",
-  observability: "activity", metrics: "gauge", runs: "play", websocket: "bolt", ag_ui: "external", remote_tools: "wrench",
+  observability: "activity", metrics: "gauge", audit: "file", runs: "play", websocket: "bolt", ag_ui: "external", remote_tools: "wrench",
   media: "file", checkpointer: "layers", id_generator: "key", thread_name_generator: "file", store: "box",
   injectq: "arrows", test: "flask", evaluation: "target",
 };

config-editor-ui/src/pages.jsx

@@ -8,6 +8,7 @@ import { Callout, Disclosure, Icon, IssueCallout, Segmented } from "./ui.jsx";
 import {
   AuthWidget, CacheStoreWidget, ChoiceWidget, RemoteToolsWidget,
 } from "./widgets.jsx";
+import { AuditWidget, auditSummary } from "./audit.jsx";
 import { MetricsWidget, TracingWidget, metricsSummary, traceSummary } from "./telemetry.jsx";
 
 const Heading = ({ children, code }) => (
@@ -72,7 +73,7 @@ function GetRunning() {
   const core = section("core");
   const ready = isAgentReady();
   const general = core ? issuesFor("core").filter((i) => !core.fields.some((f) => f.path.join(".") === i.path)) : [];
-  const rows = [["auth", authSummary()], ["cache_store", cacheSummary()], ["observability", traceSummary(offText)], ["metrics", metricsSummary(offText)]];
+  const rows = [["auth", authSummary()], ["cache_store", cacheSummary()], ["observability", traceSummary(offText)], ["metrics", metricsSummary(offText)], ["audit", auditSummary(offText)]];
   const all = store.nav.flatMap((g) => g.items).filter((id) => id !== "get-running");
   return (
     <>
@@ -142,6 +143,7 @@ function SectionWidget({ sec }) {
   if (sec.widget === "remote_tools") return <RemoteToolsWidget />;
   if (sec.widget === "observability") return <TracingWidget />;
   if (sec.widget === "metrics") return <MetricsWidget />;
+  if (sec.widget === "audit") return <AuditWidget />;
   if (sec.widget === "cache_store") return <CacheStoreWidget sec={sec} />;
   if (sec.widget === "id_generator" || sec.widget === "thread_name_generator") return <ChoiceWidget sec={sec} />;
   return null;

config-editor-ui/src/store.js

@@ -7,7 +7,7 @@
 // are pages of their own rather than schema sections.
 export const NAV = [
   { label: null, items: ["get-running"] },
-  { label: "Before production", items: ["auth", "cache_store", "observability", "metrics"] },
+  { label: "Before production", items: ["auth", "cache_store", "observability", "metrics", "audit"] },
   { label: "Runs and streaming", items: ["runs", "websocket", "ag_ui", "remote_tools", "media"] },
   { label: "Project wiring", items: ["checkpointer", "id_generator", "thread_name_generator", "store", "injectq"] },
   { label: "Developer tools", items: ["test", "evaluation"] },
@@ -156,6 +156,7 @@ export const sectionChanged = (sec) => {
   if (sec.id === "core") return false;
   if (sec.id === "auth") return store.config.auth != null;
   if (sec.id === "metrics") return store.config.metrics?.enabled === true;
+  if (sec.id === "audit") return store.config.audit?.enabled === true;
   return sec.optional ? isOn(sec) : sec.fields.some(fieldChanged);
 };
 // Highest-severity issue of a section, or null.

tenxgraph_api/__init__.py

@@ -10,6 +10,7 @@
 # Lets expose few things the user suppose to use
 from tenxgraph.core.authz import Identity
 
+from .src.app.core.audit import AuditEvent, BaseAuditSink
 from .src.app.core.auth.base_auth import BaseAuth
 from .src.app.core.auth.permissions import RequireIdentity, RequirePermission
 from .src.app.core.cache_store import (
@@ -25,6 +26,8 @@
 
 
 __all__ = [
+    "AuditEvent",
+    "BaseAuditSink",
     "BaseAuth",
     "BaseCacheStore",
     "Identity",

tenxgraph_api/cli/commands/audit.py

@@ -24,6 +24,7 @@
 
 import importlib
 import json
+import os
 import socket
 import sys
 from collections.abc import Callable
@@ -57,7 +58,7 @@ class Diagnostic:
 class AuditCommand(BaseCommand):
     """Audit the local CLI, core package, project config, and default port.
 
-    Runs eight checks in a fixed order and reports them twice: live, through the
+    Runs nine checks in a fixed order and reports them twice: live, through the
     step timeline, and again as a summary table once every check has finished.
 
     ============  ==========================================================
@@ -78,6 +79,8 @@ class AuditCommand(BaseCommand):
     security      The settings in ``10xgraph.json`` that matter before production
                   (auth, authorization, rate limiting, token revocation), as
                   warnings only: they are the developer's choices, never failures.
+    audit log     Warns when MODE=production and ``audit.enabled`` is not true
+                  in ``10xgraph.json``.
     port          The default API port is free to bind.
     ============  ==========================================================
     """
@@ -108,6 +111,7 @@ def execute(self, *, config: str = DEFAULT_CONFIG_FILE, **kwargs: Any) -> int:
             ("config", "Project configuration", lambda: self._config_check(config)),
             ("auth", "Auth configuration", lambda: self._auth_check(config)),
             ("security", "Security posture", lambda: self._security_check(config)),
+            ("audit_log", "Audit log", lambda: self._audit_log_check(config)),
             ("port", f"Port {DEFAULT_PORT}", lambda: self._port_check(DEFAULT_PORT)),
         )
 
@@ -253,6 +257,21 @@ def _security_check(config: str = DEFAULT_CONFIG_FILE) -> Diagnostic:
             "Security", "warn", "; ".join(f"{i['path']}: {i['message']}" for i in issues)
         )
 
+    @staticmethod
+    def _audit_log_check(config: str = DEFAULT_CONFIG_FILE) -> Diagnostic:
+        """Warn when production runs without the audit log (reads the JSON and MODE only)."""
+        if os.environ.get("MODE", "").lower() != "production":
+            return Diagnostic("Audit log", "pass", "not checked outside MODE=production")
+        config_path = resolve_default_config(config).resolve()
+        try:
+            data = json.loads(config_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return Diagnostic("Audit log", "pass", "no readable config to check")
+        section = data.get("audit") if isinstance(data, dict) else None
+        if isinstance(section, dict) and section.get("enabled") is True:
+            return Diagnostic("Audit log", "pass", "on")
+        return Diagnostic("Audit log", "warn", "audit log off in production")
+
     @staticmethod
     def _port_check(port: int) -> Diagnostic:
         """Warn when ``port`` on loopback already has a listener.

tenxgraph_api/cli/commands/init.py

@@ -291,6 +291,7 @@ def _print_summary(self, context: dict) -> None:
 
             store = context.get("cache_store") or "memory"
             rows.append(("Cache store", "Redis" if store == "redis" else "Memory"))
+            rows.append(("Audit log", "On"))
             rows.append(("ID generator", str(context.get("id_generator", "uuid"))))
             if context.get("rate_limit"):
                 rl_by = {"global": "Global", "user": "Per user"}.get(

tenxgraph_api/cli/commands/init_context.py

@@ -267,6 +267,9 @@ def build_config(context: dict, is_prod: bool) -> dict:
         cache_store["token_revocation"] = True
     config["cache_store"] = cache_store
 
+    # Who accessed or changed which data, as JSON lines on stdout for the log pipeline.
+    config["audit"] = {"enabled": True, "sinks": ["log"], "include_reads": True}
+
     worker = WORKERS.get(context.get("worker", "none"))
     if worker:
         config["worker"] = {