feat: add auditable logging with user_id/org_id correlation

Implement structured audit logging that correlates Gemini user sessions
with Red Hat API requests.  Every JSON log record now includes user_id,
org_id, order_id, and a per-request UUID4 request_id, providing a full
audit trail of data lineage independent of ephemeral session storage.

- Add _request_user_id, _request_org_id, _request_id contextvars to auth
  middleware with getter functions
- Create AuditContextFilter logging filter that injects audit fields into
  every log record from contextvars
- Update JSON log format to include user_id, org_id, order_id, request_id
- Enhance AgentLoggingPlugin with event_type classification and
  data_source tracking for MCP tool calls
- Add mcp_jwt_forwarded audit event to MCP header provider
- Document audit logging in .env.example, docs/configuration.md,
  deploy/cloudrun/README.md, service.yaml, and podman configmap

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: luis5tb

.env.example

@@ -154,6 +154,21 @@ LOG_FORMAT=json
 #   detailed - Also logs tool arguments and truncated results (may contain user data)
 AGENT_LOGGING_DETAIL=basic
 
+# Audit logging (automatic when LOG_FORMAT=json):
+# JSON log records automatically include user_id, org_id, order_id, and
+# request_id fields for every log entry.  These fields are populated from
+# the authenticated user's JWT token and provide a full audit trail of
+# data lineage — proving that information shown to the user was authorized
+# and retrieved from a verified Red Hat source.
+#
+# Each agent lifecycle event is tagged with an event_type:
+#   request_authenticated, agent_run_started, agent_run_completed,
+#   llm_call_started, llm_call_completed, tool_call_started,
+#   tool_call_completed, mcp_jwt_forwarded
+#
+# Tool calls include a data_source field identifying the Red Hat Insights
+# MCP tool that retrieved the data.  No additional configuration is required.
+
 # -----------------------------------------------------------------------------
 # Development/Debug Settings
 # -----------------------------------------------------------------------------

deploy/cloudrun/README.md

@@ -1309,6 +1309,43 @@ The test script at `scripts/test_deployed_dcr.py` is configurable via environmen
 | `TEST_ACCOUNT_ID` | `test-procurement-account-001` | Procurement account ID |
 | `TEST_REDIRECT_URIS` | `https://gemini.google.com/callback` | Comma-separated redirect URIs |
 
+## Audit Logging
+
+The agent automatically produces structured audit logs that correlate each user session with Red Hat API requests. When `LOG_FORMAT=json` (the default in Cloud Run), every log record includes:
+
+- **`user_id`** — authenticated user (JWT `sub` claim)
+- **`org_id`** — Red Hat organization (JWT `org_id` claim)
+- **`order_id`** — Google Cloud Marketplace order
+- **`request_id`** — UUID4 correlation ID (unique per request)
+
+Each agent lifecycle event carries an `event_type` tag (`request_authenticated`, `agent_run_started`, `tool_call_completed`, `mcp_jwt_forwarded`, etc.) and tool calls include a `data_source` field identifying which Red Hat Insights MCP tool retrieved the data.
+
+This provides a full data lineage audit trail: every piece of information disclosed by the agent can be traced back to a specific authenticated user prompt and a verified Red Hat Insights data source. These persistent logs are independent of the ephemeral ADK session storage.
+
+### Querying Audit Logs
+
+Cloud Logging automatically parses JSON log fields. Example queries:
+
+```bash
+# All actions by a specific user
+gcloud logging read 'jsonPayload.user_id="<user-id>"' \
+  --project=$GOOGLE_CLOUD_PROJECT --limit=50
+
+# All events in a single request (correlation)
+gcloud logging read 'jsonPayload.request_id="<request-id>"' \
+  --project=$GOOGLE_CLOUD_PROJECT
+
+# All MCP data access for an organization
+gcloud logging read 'jsonPayload.org_id="<org-id>" AND jsonPayload.message=~"mcp_jwt_forwarded"' \
+  --project=$GOOGLE_CLOUD_PROJECT
+
+# All tool calls with data source tracking
+gcloud logging read 'jsonPayload.message=~"tool_call_completed"' \
+  --project=$GOOGLE_CLOUD_PROJECT --limit=20
+```
+
+No additional configuration is required — audit logging is automatically active when `LOG_FORMAT=json`.
+
 ## Monitoring
 
 View metrics in Google Cloud Console:

deploy/cloudrun/service.yaml

@@ -81,6 +81,10 @@ spec:
             # Updated automatically by deploy.sh after deployment.
             - name: MARKETPLACE_HANDLER_URL
               value: "https://marketplace-handler.example.com"
+            # Logging and Audit Trail
+            # LOG_FORMAT=json enables structured audit logging with user_id,
+            # org_id, order_id, and request_id in every log record.  Cloud
+            # Logging parses these fields automatically for querying.
             - name: LOG_LEVEL
               value: "INFO"
             - name: LOG_FORMAT

deploy/podman/lightspeed-agent-configmap.yaml

@@ -73,7 +73,9 @@ data:
   # Set SERVICE_CONTROL_SERVICE_NAME in secrets for production
   SERVICE_CONTROL_ENABLED: "false"
 
-  # Logging
+  # Logging and Audit Trail
+  # LOG_FORMAT=json enables structured audit logging with user_id,
+  # org_id, order_id, and request_id in every log record.
   LOG_LEVEL: "DEBUG"
   LOG_FORMAT: "json"
   AGENT_LOGGING_DETAIL: "detailed"

docs/configuration.md

@@ -249,6 +249,55 @@ LOG_FORMAT=text  # Human-readable for development
 AGENT_LOGGING_DETAIL=detailed  # Include tool args/results in logs
 ```
 
+#### Audit Logging
+
+When `LOG_FORMAT=json` (the default), every log record automatically includes audit context fields:
+
+| Field | Source | Description |
+|-------|--------|-------------|
+| `user_id` | JWT `sub` claim | Authenticated user identifier |
+| `org_id` | JWT `org_id` claim | Red Hat organization identifier |
+| `order_id` | DCR client lookup | Google Cloud Marketplace order |
+| `request_id` | Generated UUID4 | Per-request correlation ID |
+
+These fields enable:
+- **Request correlation** — all events in a single request share the same `request_id`
+- **User audit** — filter by `user_id` to trace all actions by a specific user
+- **Organization audit** — filter by `org_id` for organization-level auditing
+- **Data lineage** — `tool_call_completed` events include `data_source=<mcp_tool>`, and `mcp_jwt_forwarded` events prove data was retrieved using the user's authorized JWT
+
+Each agent lifecycle event is tagged with an `event_type` in the log message:
+
+| Event Type | Description |
+|------------|-------------|
+| `request_authenticated` | User JWT validated, user_id and org_id extracted |
+| `agent_run_started` | ADK agent invocation started |
+| `agent_run_completed` | ADK agent invocation finished |
+| `llm_call_started` | Gemini LLM call initiated |
+| `llm_call_completed` | Gemini LLM call finished (includes token counts) |
+| `tool_call_started` | MCP tool call initiated |
+| `tool_call_completed` | MCP tool call finished (includes `data_source`) |
+| `mcp_jwt_forwarded` | User JWT forwarded to MCP sidecar for Red Hat API auth |
+
+**Example JSON log line:**
+
+```json
+{"time": "2025-01-15 10:30:45", "level": "INFO", "logger": "lightspeed_agent.api.a2a.logging_plugin", "message": "Tool call completed (event_type=tool_call_completed, tool=get_advisories, data_source=get_advisories, ...)", "user_id": "user-42", "org_id": "org-7", "order_id": "order-99", "request_id": "abc-123-def-456"}
+```
+
+On Cloud Run, these JSON logs are automatically parsed by Cloud Logging and can be queried with:
+
+```bash
+# Find all actions by a specific user
+gcloud logging read 'jsonPayload.user_id="user-42"' --project=$GOOGLE_CLOUD_PROJECT
+
+# Find all tool calls in a specific request
+gcloud logging read 'jsonPayload.request_id="abc-123" AND jsonPayload.message=~"tool_call"' --project=$GOOGLE_CLOUD_PROJECT
+
+# Audit all MCP data access for an organization
+gcloud logging read 'jsonPayload.org_id="org-7" AND jsonPayload.message=~"mcp_jwt_forwarded"' --project=$GOOGLE_CLOUD_PROJECT
+```
+
 ### Development Settings
 
 | Variable | Default | Description |

src/lightspeed_agent/api/a2a/logging_plugin.py

@@ -1,8 +1,11 @@
-"""Agent execution logging plugin.
+"""Agent execution logging plugin with audit context.
 
 Logs agent lifecycle events (tool calls, LLM invocations, run start/end)
-at INFO level for operational visibility. Controlled by the
-AGENT_LOGGING_DETAIL setting:
+at INFO level for operational visibility. Each log entry includes an
+``event_type`` classification and audit context (user_id, org_id,
+order_id, request_id) for data lineage tracing.
+
+Controlled by the AGENT_LOGGING_DETAIL setting:
   - "basic": logs tool names, token counts, and lifecycle events
   - "detailed": also logs tool arguments and truncated results
 """
@@ -18,6 +21,12 @@
 from google.adk.tools.base_tool import BaseTool
 from google.adk.tools.tool_context import ToolContext
 
+from lightspeed_agent.auth.middleware import (
+    get_request_id,
+    get_request_order_id,
+    get_request_org_id,
+    get_request_user_id,
+)
 from lightspeed_agent.config import get_settings
 
 logger = logging.getLogger(__name__)
@@ -34,32 +43,46 @@ def _truncate(value: Any, max_length: int = _MAX_RESULT_LENGTH) -> str:
 
 
 class AgentLoggingPlugin(BasePlugin):
-    """ADK plugin that logs agent execution events at INFO level."""
+    """ADK plugin that logs agent execution events with audit context."""
 
     def __init__(self) -> None:
         super().__init__(name="agent_logging")
 
     def _is_detailed(self) -> bool:
         return get_settings().agent_logging_detail == "detailed"
 
+    @staticmethod
+    def _audit_fields() -> str:
+        """Build audit context string for log messages."""
+        return (
+            f"user_id={get_request_user_id()}, "
+            f"org_id={get_request_org_id()}, "
+            f"order_id={get_request_order_id()}, "
+            f"request_id={get_request_id()}"
+        )
+
     # -- run lifecycle --------------------------------------------------------
 
     async def before_run_callback(
         self, *, invocation_context: InvocationContext
     ) -> None:
         logger.info(
-            "Agent run started (invocation_id=%s, agent=%s)",
+            "Agent run started "
+            "(event_type=agent_run_started, invocation_id=%s, agent=%s, %s)",
             invocation_context.invocation_id,
             invocation_context.agent.name,
+            self._audit_fields(),
         )
         return None
 
     async def after_run_callback(
         self, *, invocation_context: InvocationContext
     ) -> None:
         logger.info(
-            "Agent run completed (invocation_id=%s)",
+            "Agent run completed "
+            "(event_type=agent_run_completed, invocation_id=%s, %s)",
             invocation_context.invocation_id,
+            self._audit_fields(),
         )
         return None
 
@@ -68,7 +91,11 @@ async def after_run_callback(
     async def before_model_callback(
         self, *, callback_context: CallbackContext, llm_request: LlmRequest
     ) -> Any | None:
-        logger.info("LLM call started (agent=%s)", callback_context.agent_name)
+        logger.info(
+            "LLM call started (event_type=llm_call_started, agent=%s, %s)",
+            callback_context.agent_name,
+            self._audit_fields(),
+        )
         return None
 
     async def after_model_callback(
@@ -87,17 +114,24 @@ async def after_model_callback(
         model_version = llm_response.model_version if llm_response else None
 
         logger.info(
-            "LLM call completed (input_tokens=%d, output_tokens=%d, model=%s)",
+            "LLM call completed "
+            "(event_type=llm_call_completed, input_tokens=%d, output_tokens=%d, "
+            "model=%s, %s)",
             input_tokens,
             output_tokens,
             model_version,
+            self._audit_fields(),
         )
         return None
 
     async def on_model_error_callback(
         self, *, callback_context: CallbackContext, llm_request: LlmRequest, error: Exception
     ) -> LlmResponse | None:
-        logger.error("LLM call failed: %s", error)
+        logger.error(
+            "LLM call failed (event_type=llm_call_failed, error=%s, %s)",
+            error,
+            self._audit_fields(),
+        )
         return None
 
     # -- tool callbacks -------------------------------------------------------
@@ -112,12 +146,19 @@ async def before_tool_callback(
         tool_name = getattr(tool, "name", type(tool).__name__)
         if self._is_detailed():
             logger.info(
-                "Tool call started (tool=%s, args=%s)",
+                "Tool call started "
+                "(event_type=tool_call_started, tool=%s, args=%s, %s)",
                 tool_name,
                 _truncate(tool_args),
+                self._audit_fields(),
             )
         else:
-            logger.info("Tool call started (tool=%s)", tool_name)
+            logger.info(
+                "Tool call started "
+                "(event_type=tool_call_started, tool=%s, %s)",
+                tool_name,
+                self._audit_fields(),
+            )
         return None
 
     async def after_tool_callback(
@@ -131,12 +172,22 @@ async def after_tool_callback(
         tool_name = getattr(tool, "name", type(tool).__name__)
         if self._is_detailed():
             logger.info(
-                "Tool call completed (tool=%s, result=%s)",
+                "Tool call completed "
+                "(event_type=tool_call_completed, tool=%s, data_source=%s, "
+                "result=%s, %s)",
+                tool_name,
                 tool_name,
                 _truncate(result),
+                self._audit_fields(),
             )
         else:
-            logger.info("Tool call completed (tool=%s)", tool_name)
+            logger.info(
+                "Tool call completed "
+                "(event_type=tool_call_completed, tool=%s, data_source=%s, %s)",
+                tool_name,
+                tool_name,
+                self._audit_fields(),
+            )
         return None
 
     async def on_tool_error_callback(
@@ -148,5 +199,11 @@ async def on_tool_error_callback(
         error: Exception,
     ) -> dict[str, Any] | None:
         tool_name = getattr(tool, "name", type(tool).__name__)
-        logger.error("Tool call failed (tool=%s): %s", tool_name, error)
+        logger.error(
+            "Tool call failed "
+            "(event_type=tool_call_failed, tool=%s, error=%s, %s)",
+            tool_name,
+            error,
+            self._audit_fields(),
+        )
         return None

src/lightspeed_agent/auth/__init__.py

@@ -17,7 +17,12 @@
     TokenValidationError,
     get_token_introspector,
 )
-from lightspeed_agent.auth.middleware import AuthenticationMiddleware
+from lightspeed_agent.auth.middleware import (
+    AuthenticationMiddleware,
+    get_request_id,
+    get_request_org_id,
+    get_request_user_id,
+)
 from lightspeed_agent.auth.models import (
     AuthenticatedUser,
     JWTClaims,
@@ -36,6 +41,9 @@
     "get_token_introspector",
     # Middleware
     "AuthenticationMiddleware",
+    "get_request_id",
+    "get_request_org_id",
+    "get_request_user_id",
     # Models
     "AuthenticatedUser",
     "JWTClaims",