feat(obd_ai): add policy guardrails and audit logging

docs/LLM Tool Server.md

@@ -17,6 +17,7 @@ This transport is function-calling-friendly JSON over HTTP and reuses the same c
 - `basic_health_scan`
 
 These tools remain read-only. The server does **not** expose raw OBD command construction or write operations.
+Sensitive or destructive actions such as `CLEAR_DTC` are covered by the policy layer, which returns structured approval-required or blocked responses instead of executing them.
 
 ## Start the server
 
@@ -81,4 +82,6 @@ Equivalent convenience route for directly calling a specific tool.
 
 - Session lifecycle is managed by the underlying tool surface.
 - Call `connect_vehicle` before issuing read operations.
+- The tool surface emits structured audit events for tool invocations and policy decisions.
+- Future motion, ignition, and vehicle-state guards plug into the policy metadata without exposing raw commands.
 - Error payloads are returned in structured JSON and preserve existing `obd_ai` error codes when possible.

obd_ai/__init__.py

@@ -1,10 +1,24 @@
 """Safe, high-level interfaces for LLM/MCP-style OBD access."""
 
+from .audit import AuditEvent, InMemoryAuditSink, LoggerAuditSink
 from .catalog import (
     ApprovedCommand,
     ApprovedCommandCatalog,
     DEFAULT_APPROVED_COMMAND_CATALOG,
 )
+from .policy import (
+    DEFAULT_POLICY_CATALOG,
+    DECISION_ALLOW,
+    DECISION_DENY,
+    DECISION_REQUIRE_APPROVAL,
+    OBDAIPolicyEngine,
+    POLICY_CLASS_BLOCKED,
+    POLICY_CLASS_DESTRUCTIVE,
+    POLICY_CLASS_READ_ONLY,
+    POLICY_CLASS_SENSITIVE,
+    PolicyAction,
+    PolicyDecision,
+)
 from .serializers import (
     serialize_approved_command,
     serialize_approved_command_catalog,
@@ -19,10 +33,24 @@
 __all__ = [
     "ApprovedCommand",
     "ApprovedCommandCatalog",
+    "AuditEvent",
     "DEFAULT_APPROVED_COMMAND_CATALOG",
+    "DEFAULT_POLICY_CATALOG",
+    "DECISION_ALLOW",
+    "DECISION_DENY",
+    "DECISION_REQUIRE_APPROVAL",
+    "InMemoryAuditSink",
+    "LoggerAuditSink",
+    "OBDAIPolicyEngine",
     "OBDAISession",
     "OBDAISessionManager",
     "OBDAIReadOnlyToolSurface",
+    "POLICY_CLASS_BLOCKED",
+    "POLICY_CLASS_DESTRUCTIVE",
+    "POLICY_CLASS_READ_ONLY",
+    "POLICY_CLASS_SENSITIVE",
+    "PolicyAction",
+    "PolicyDecision",
     "serialize_approved_command",
     "serialize_approved_command_catalog",
     "serialize_connection_metadata",

obd_ai/audit.py

@@ -0,0 +1,91 @@
+"""Audit logging primitives for the OBD AI tool surface."""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, Mapping, Optional, Protocol
+
+
+@dataclass(frozen=True)
+class AuditEvent:
+    """Structured audit event emitted by the OBD AI layer."""
+
+    event_type: str
+    tool: str
+    requested_action: str
+    request: Dict[str, Any]
+    decision: Optional[str] = None
+    result: Optional[str] = None
+    error_code: Optional[str] = None
+    policy: Optional[Dict[str, Any]] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    occurred_at: str = field(
+        default_factory=lambda: datetime.now(timezone.utc).isoformat()
+    )
+
+    def to_dict(self) -> Dict[str, Any]:
+        payload = {
+            "event_type": self.event_type,
+            "tool": self.tool,
+            "requested_action": self.requested_action,
+            "request": dict(self.request),
+            "decision": self.decision,
+            "result": self.result,
+            "error_code": self.error_code,
+            "policy": dict(self.policy) if self.policy is not None else None,
+            "metadata": dict(self.metadata),
+            "occurred_at": self.occurred_at,
+        }
+        return payload
+
+
+class AuditSink(Protocol):
+    def record(self, event: AuditEvent) -> None:
+        """Persist or forward an audit event."""
+
+
+class LoggerAuditSink:
+    """Default audit sink that writes structured events to ``logging``."""
+
+    def __init__(self, logger: Optional[logging.Logger] = None):
+        self._logger = logger or logging.getLogger("obd_ai.audit")
+
+    def record(self, event: AuditEvent) -> None:
+        self._logger.info("obd_ai_audit %s", json.dumps(event.to_dict(), sort_keys=True))
+
+
+class InMemoryAuditSink:
+    """Test-friendly audit sink that stores serialized events in memory."""
+
+    def __init__(self):
+        self.events = []
+
+    def record(self, event: AuditEvent) -> None:
+        self.events.append(event.to_dict())
+
+
+def build_audit_event(
+    event_type: str,
+    tool: str,
+    requested_action: str,
+    request: Optional[Mapping[str, Any]] = None,
+    decision: Optional[str] = None,
+    result: Optional[str] = None,
+    error_code: Optional[str] = None,
+    policy: Optional[Mapping[str, Any]] = None,
+    metadata: Optional[Mapping[str, Any]] = None,
+) -> AuditEvent:
+    return AuditEvent(
+        event_type=event_type,
+        tool=tool,
+        requested_action=requested_action,
+        request=dict(request or {}),
+        decision=decision,
+        result=result,
+        error_code=error_code,
+        policy=dict(policy) if policy is not None else None,
+        metadata=dict(metadata or {}),
+    )

obd_ai/catalog.py

@@ -18,6 +18,9 @@ class ApprovedCommand:
     public_name: str
     category: str
     description: str
+    policy_class: str
+    approval_required: bool
+    guard_tags: Tuple[str, ...]
     obd_command_name: str
     obd_command: OBDCommand
 
@@ -29,6 +32,9 @@ class _CommandDefinition:
     category: str
     description: str
     obd_command_name: str
+    policy_class: str = "read_only"
+    approval_required: bool = False
+    guard_tags: Tuple[str, ...] = ()
 
 
 class ApprovedCommandCatalog:
@@ -54,6 +60,9 @@ def __init__(self, definitions: Iterable[_CommandDefinition]):
                 public_name=definition.public_name,
                 category=definition.category,
                 description=definition.description,
+                policy_class=definition.policy_class,
+                approval_required=definition.approval_required,
+                guard_tags=definition.guard_tags,
                 obd_command_name=definition.obd_command_name,
                 obd_command=obd_command,
             )