An agent deleted a production database table at 2 AM. The audit log showed: "Agent executed DROP TABLE users at 2024-11-15T02:17:43Z." That's what happened. But we had no idea why. What context made the agent decide to drop the table? What was in the conversation history? What documents had it read? What API responses influenced the decision? The audit log was useless for investigation.
Traditional audit logging captures actions—who did what, when. For deterministic systems, this is sufficient. You can trace code execution, replay requests, and understand causation. An admin executed a command. A script ran a procedure. The action itself explains the intent.
For agents, actions don't explain intent because intent emerges from context we don't control. The agent didn't spontaneously decide to drop a table. Something in its context—a prompt, a document, an API response—triggered that decision. Without logging what context influenced the decision, the audit trail is incomplete.
Post-incident investigation becomes impossible. Did the agent misinterpret a legitimate user request? Was it a prompt injection attack? Did it read a poisoned document? We can't answer these questions because the audit log only captured the action, not the context that caused it.
What we need is context provenance logging: capturing not just what agents do, but why they do it. Which pieces of context were present when the decision was made? Which documents influenced the LLM's reasoning? What was the prompt that triggered this tool call? Context logging transforms audit trails from action records into decision explanations.
This is expensive—context can be megabytes per decision. But it's necessary for agent systems where decisions are non-deterministic and investigation requires reconstructing the reasoning process that led to problematic actions.
The Context Provenance Problem
Traditional audit logging assumes deterministic systems where causation is traceable through code paths. An admin runs DROP TABLE. The causation is clear: the admin intended this action and executed it. Investigation focuses on authorization: was the admin authorized? Logs show who, what, when. That's sufficient.
Agents break this model. An agent executes DROP TABLE. The causation isn't clear. The agent didn't intend anything—it responded to inputs based on probabilistic reasoning. Investigation needs to answer: what inputs caused this response? Traditional logs can't answer this.
The fundamental difference: deterministic systems have intent, agents have context.
For deterministic systems:
- Action implies intent
- Intent is traceable to actor
- Authorization question: "Should this actor perform this action?"
- Audit answer: "This actor performed this action at this time"
For agent systems:
- Action implies context influenced reasoning
- Context is distributed across multiple sources
- Causation question: "What context caused this decision?"
- Audit answer: "This agent performed this action after processing this context"
Without context, agent audit logs are "what" without "why." You know the agent dropped a table. You don't know if it was following a legitimate user request, misinterpreting instructions, or responding to a prompt injection attack.
The correct mental model: Audit trails as decision explanations
Traditional auditing: Record who did what when
Context provenance auditing: Record what context led to what decision leading to what action
The shift is from action logging to decision logging. Instead of just recording that a tool was called, record the decision process: what prompted the agent to call this tool, what context was available, what reasoning led to this choice.
The invariant to maintain:
For every agent action, the audit trail must contain sufficient context to reconstruct why that action was chosen. Not just "what happened" but "what caused it to happen."
The key insight:
Agents are context processors. Their outputs are functions of their inputs. To understand outputs (actions), you must log inputs (context). This is obvious in theory but challenging in practice because context is large, distributed, and partially untrusted.
Context Provenance Architecture
A context provenance system captures context sources, tracks which context influenced which decisions, and maintains queryable logs linking actions back to causal context.
Agentic AI: Context Provenance Architecture
Component responsibilities:
Context Collector: Captures all context sources before they enter agent processing. Assigns unique IDs to each context piece.
Context Registry: Maintains mapping of context IDs to sources. Enables tracking which context came from where.
Context Store: Persistent storage of full context content and metadata. Separate from audit database due to size.
Agent Decision Layer: Makes decisions based on collected context. All context is tracked and referenced.
Decision Logger: Records decisions with references to influencing context. Links actions to context provenance.
Audit Record: Structured log entry containing action details, context references, decision reasoning, and timestamp.
Query Interface: Enables investigation queries linking actions back to causal context.
Incident Investigation: Uses audit records and context store to reconstruct decision processes and identify root causes.
Key architectural properties:
Context ID assignment: Every piece of context gets unique ID before processing. Enables precise provenance tracking.
Separation of concerns: Audit database stores references and metadata. Context store holds full content. Prevents audit database bloat.
Decision-centric logging: Logs focus on decisions (why this tool was called) not just actions (tool was called).
Queryable provenance: Can query "what context led to action X" or "what actions resulted from context Y."
Temporal reconstruction: Can reconstruct complete context window at any point in agent execution.
Implementation: Building Context Provenance
Here's what context provenance logging looks like in production, based on systems I've built for incident investigation.
Context Tracking and Provenance
from dataclasses import dataclassfrom typing import List, Dict, Any, Optionalfrom datetime import datetimeimport hashlibimport jsonimport uuid@dataclassclass ContextSource: """Individual piece of context with provenance.""" context_id: str source_type: str # "user_prompt", "document", "api_response", "file", "system" content: str content_hash: str metadata: Dict[str, Any] timestamp: datetime trust_level: float # 0.0-1.0, how much we trust this source class ContextCollector: """ Collects and tracks all context entering agent processing. """ def __init__(self): self.context_store = ContextStore() self.context_registry = {} def collect( self, source_type: str, content: str, metadata: Dict[str, Any], trust_level: float = 0.5 ) -> str: """ Collect context and assign ID for provenance tracking. """ # Generate unique context ID context_id = self._generate_context_id() # Hash content for integrity content_hash = hashlib.sha256(content.encode()).hexdigest() # Create context source context = ContextSource( context_id=context_id, source_type=source_type, content=content, content_hash=content_hash, metadata=metadata, timestamp=datetime.utcnow(), trust_level=trust_level ) # Store full context self.context_store.store(context) # Register in memory for fast lookups self.context_registry[context_id] = context return context_id def get_context(self, context_id: str) -> Optional[ContextSource]: """Retrieve context by ID.""" if context_id in self.context_registry: return self.context_registry[context_id] return self.context_store.retrieve(context_id) def _generate_context_id(self) -> str: """Generate unique context identifier.""" return f"ctx_{uuid.uuid4().hex}"@dataclassclass DecisionRecord: """ Record of agent decision with full context provenance. """ decision_id: str agent_id: str timestamp: datetime # The decision decision_type: str # "tool_call", "response_generation", etc. decision_details: Dict[str, Any] # Context provenance context_ids: List[str] # Which context influenced this decision primary_context_id: Optional[str] # Most influential context # Reasoning (if available) reasoning: Optional[str] # Outcome action_taken: str action_parameters: Dict[str, Any] action_result: Optional[str]class DecisionLogger: """ Logs agent decisions with context provenance. """ def __init__(self): self.audit_db = AuditDatabase() self.context_collector = ContextCollector() def log_decision( self, agent_id: str, decision_type: str, decision_details: Dict[str, Any], context_ids: List[str], action_taken: str, action_parameters: Dict[str, Any], reasoning: Optional[str] = None, primary_context_id: Optional[str] = None ) -> str: """ Log decision with full context provenance. """ decision_id = self._generate_decision_id() record = DecisionRecord( decision_id=decision_id, agent_id=agent_id, timestamp=datetime.utcnow(), decision_type=decision_type, decision_details=decision_details, context_ids=context_ids, primary_context_id=primary_context_id, reasoning=reasoning, action_taken=action_taken, action_parameters=action_parameters, action_result=None # Will be updated later ) # Persist to audit database self.audit_db.store_decision(record) return decision_id def update_decision_outcome( self, decision_id: str, result: str ): """Update decision record with execution result.""" self.audit_db.update_decision_result(decision_id, result) def _generate_decision_id(self) -> str: """Generate unique decision identifier.""" return f"dec_{uuid.uuid4().hex}"class ContextProvenanceAgent: """ Agent that logs context provenance for all decisions. """ def __init__(self, agent_id: str): self.agent_id = agent_id self.context_collector = ContextCollector() self.decision_logger = DecisionLogger() self.active_context_ids = [] def process_user_input(self, user_prompt: str) -> str: """Process user input with full context logging.""" # Collect user prompt as context prompt_context_id = self.context_collector.collect( source_type="user_prompt", content=user_prompt, metadata={"user_id": "...", "session_id": "..."}, trust_level=0.4 # User input is untrusted ) self.active_context_ids.append(prompt_context_id) # Agent processing happens here # For this example, assume agent decides to call a tool decision_id = self.decision_logger.log_decision( agent_id=self.agent_id, decision_type="tool_call", decision_details={"tool": "database_query"}, context_ids=self.active_context_ids, primary_context_id=prompt_context_id, action_taken="query_database", action_parameters={"table": "users", "query": "SELECT * FROM users"}, reasoning="User requested user information" ) # Execute tool result = self._execute_tool("query_database", {"table": "users"}) # Update decision with result self.decision_logger.update_decision_outcome(decision_id, str(result)) # Collect tool result as new context result_context_id = self.context_collector.collect( source_type="tool_response", content=str(result), metadata={"tool": "database_query", "decision_id": decision_id}, trust_level=0.8 # Tool responses are moderately trusted ) self.active_context_ids.append(result_context_id) return str(result) def add_document_context(self, document_content: str, source: str): """Add document to agent context with provenance tracking.""" doc_context_id = self.context_collector.collect( source_type="document", content=document_content, metadata={"source": source}, trust_level=0.6 # Documents are somewhat trusted ) self.active_context_ids.append(doc_context_id) def _execute_tool(self, tool_name: str, parameters: Dict) -> Any: """Execute tool (placeholder).""" return {"status": "success", "data": [...]}class IncidentInvestigator: """ Investigate incidents using context provenance. """ def __init__(self): self.audit_db = AuditDatabase() self.context_store = ContextStore() def investigate_action(self, action_description: str) -> Dict[str, Any]: """ Investigate why agent took a specific action. """ # Find decision records matching action decisions = self.audit_db.query_decisions( action_filter=action_description ) if not decisions: return {"error": "No decisions found matching action"} # For each decision, reconstruct context investigations = [] for decision in decisions: context_analysis = self._analyze_decision_context(decision) investigations.append({ "decision_id": decision.decision_id, "timestamp": decision.timestamp, "action": decision.action_taken, "context_analysis": context_analysis, "reasoning": decision.reasoning }) return {"investigations": investigations} def _analyze_decision_context( self, decision: DecisionRecord ) -> Dict[str, Any]: """ Analyze context that influenced a decision. """ # Retrieve all context that influenced this decision contexts = [ self.context_store.retrieve(ctx_id) for ctx_id in decision.context_ids ] # Categorize context by source context_by_source = {} for ctx in contexts: if ctx: if ctx.source_type not in context_by_source: context_by_source[ctx.source_type] = [] context_by_source[ctx.source_type].append({ "content_preview": ctx.content[:200], "trust_level": ctx.trust_level, "timestamp": ctx.timestamp }) # Identify primary influencer primary_context = None if decision.primary_context_id: primary_context = self.context_store.retrieve( decision.primary_context_id ) return { "context_sources": list(context_by_source.keys()), "context_by_source": context_by_source, "primary_influence": { "source_type": primary_context.source_type if primary_context else None, "content_preview": primary_context.content[:200] if primary_context else None, "trust_level": primary_context.trust_level if primary_context else None } if primary_context else None, "total_context_items": len(contexts) } def trace_context_flow( self, context_id: str ) -> List[DecisionRecord]: """ Trace all decisions influenced by specific context. """ return self.audit_db.query_decisions( context_filter=context_id ) def detect_prompt_injection( self, decision_id: str ) -> Dict[str, Any]: """ Analyze decision context for potential prompt injection attacks. """ decision = self.audit_db.get_decision(decision_id) if not decision: return {"error": "Decision not found"} # Retrieve all context contexts = [ self.context_store.retrieve(ctx_id) for ctx_id in decision.context_ids ] # Look for suspicious patterns injection_indicators = [] for ctx in contexts: if not ctx: continue content_lower = ctx.content.lower() # Check for instruction override attempts if any(phrase in content_lower for phrase in [ "ignore previous", "disregard", "new instructions", "system:", "admin mode", "override" ]): injection_indicators.append({ "context_id": ctx.context_id, "source_type": ctx.source_type, "indicator": "instruction_override", "trust_level": ctx.trust_level }) # Check for role manipulation if any(phrase in content_lower for phrase in [ "you are now", "act as", "pretend to be" ]): injection_indicators.append({ "context_id": ctx.context_id, "source_type": ctx.source_type, "indicator": "role_manipulation", "trust_level": ctx.trust_level }) # Suspicious if low-trust source contains privileged operations if ctx.trust_level < 0.5: if any(op in content_lower for op in [ "delete", "drop table", "update", "admin" ]): injection_indicators.append({ "context_id": ctx.context_id, "source_type": ctx.source_type, "indicator": "low_trust_privileged_op", "trust_level": ctx.trust_level }) return { "decision_id": decision_id, "injection_detected": len(injection_indicators) > 0, "indicators": injection_indicators, "risk_score": min(len(injection_indicators) * 0.3, 1.0) }### Storage and Retrieval Optimizationclass ContextStore: """ Optimized storage for context with tiered retention. """ def __init__(self): self.hot_storage = {} # In-memory for recent context self.db = ContextDatabase() # Persistent storage def store(self, context: ContextSource): """Store context with tiered approach.""" # Always store in hot cache self.hot_storage[context.context_id] = context # Persist to database asynchronously self._async_persist(context) # Evict old entries from hot cache self._evict_old_entries() def retrieve(self, context_id: str) -> Optional[ContextSource]: """Retrieve context from hot cache or database.""" # Check hot cache first if context_id in self.hot_storage: return self.hot_storage[context_id] # Fall back to database return self.db.get_context(context_id) def _async_persist(self, context: ContextSource): """Persist context to database asynchronously.""" # Compress large context before storage if len(context.content) > 10000: compressed = self._compress_content(context.content) self.db.store_compressed(context.context_id, compressed, context.metadata) else: self.db.store(context) def _evict_old_entries(self): """Evict entries older than 1 hour from hot cache.""" cutoff = datetime.utcnow() - timedelta(hours=1) to_evict = [ ctx_id for ctx_id, ctx in self.hot_storage.items() if ctx.timestamp < cutoff ] for ctx_id in to_evict: del self.hot_storage[ctx_id] def _compress_content(self, content: str) -> bytes: """Compress content for storage efficiency.""" import gzip return gzip.compress(content.encode('utf-8'))class AuditDatabase: """ Database for audit records with optimized queries. """ def __init__(self): self.connection = self._get_connection() def store_decision(self, record: DecisionRecord): """Store decision record with indexes for fast queries.""" query = """ INSERT INTO decision_records ( decision_id, agent_id, timestamp, decision_type, decision_details, context_ids, primary_context_id, reasoning, action_taken, action_parameters ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ self.connection.execute(query, ( record.decision_id, record.agent_id, record.timestamp.isoformat(), record.decision_type, json.dumps(record.decision_details), json.dumps(record.context_ids), record.primary_context_id, record.reasoning, record.action_taken, json.dumps(record.action_parameters) )) self.connection.commit() def query_decisions( self, action_filter: Optional[str] = None, context_filter: Optional[str] = None, agent_filter: Optional[str] = None, time_range: Optional[tuple] = None ) -> List[DecisionRecord]: """Query decisions with multiple filter options.""" conditions = [] params = [] if action_filter: conditions.append("action_taken LIKE ?") params.append(f"%{action_filter}%") if context_filter: # Search in context_ids JSON array conditions.append("context_ids LIKE ?") params.append(f"%{context_filter}%") if agent_filter: conditions.append("agent_id = ?") params.append(agent_filter) if time_range: conditions.append("timestamp BETWEEN ? AND ?") params.extend([t.isoformat() for t in time_range]) where_clause = " AND ".join(conditions) if conditions else "1=1" query = f"SELECT * FROM decision_records WHERE {where_clause} ORDER BY timestamp DESC" cursor = self.connection.execute(query, params) return [self._row_to_decision(row) for row in cursor.fetchall()] def _row_to_decision(self, row) -> DecisionRecord: """Convert database row to DecisionRecord.""" return DecisionRecord( decision_id=row[0], agent_id=row[1], timestamp=datetime.fromisoformat(row[2]), decision_type=row[3], decision_details=json.loads(row[4]), context_ids=json.loads(row[5]), primary_context_id=row[6], reasoning=row[7], action_taken=row[8], action_parameters=json.loads(row[9]), action_result=row[10] if len(row) > 10 else None ) def _get_connection(self): """Get database connection.""" import sqlite3 return sqlite3.connect('audit_trail.db')Why this works:
Context ID assignment: Every context piece gets unique ID before processing. Enables precise provenance.
Decision-centric logging: Records focus on decisions with context references, not just actions.
Multi-source tracking: Captures context from all sources (user, documents, APIs, files, system).
Trust scoring: Different context sources have different trust levels, helpful for investigation.
Bidirectional queries: Can query "what context influenced action X" or "what actions resulted from context Y."
Investigation workflows: Incident investigators can reconstruct full decision context.
Pitfalls & Failure Modes
Context provenance logging fails in production through specific patterns.
Log Volume Explosion
Context can be megabytes per decision. With thousands of decisions per hour, logs grow to terabytes monthly. Storage costs exceed compute costs. Log queries become impossibly slow.
Why it happens: Full context logging captures everything. Most of it is never queried.
Prevention: Tiered storage with compression. Recent logs (7 days) in fast storage with full context. Medium-term (90 days) compressed with sampled context. Long-term (1 year+) with metadata only, full context archived.
Context ID Reference Decay
Audit records reference context by ID. Context store has retention policy (delete after 90 days). After 90 days, context IDs in audit records point to deleted context. Investigation impossible.
Why it happens: Audit records and context store have different retention policies.
Prevention: Synchronized retention. If audit record exists, referenced context must exist. Delete audit records and context together, or archive both.
Primary Context Misidentification
Decision logger tries to identify which context was most influential. Gets it wrong. Investigator focuses on wrong context during incident analysis. Root cause is missed.
Why it happens: Determining "most influential" context requires understanding LLM reasoning, which is opaque.
Prevention: Log all context without trying to rank influence. Let investigators analyze full context set. Or, use attention weights from LLM if available.
Performance Degradation from Logging Overhead
Every decision logs context references, reasoning, action details. At scale, logging adds 50-100ms per decision. Agent latency becomes unacceptable.
Why it happens: Logging is I/O intensive. Database writes are slow.
Prevention: Asynchronous logging with buffering. Log to memory queue, background thread persists to database. Monitor queue depth to prevent data loss.
Context Privacy Leakage
Audit logs contain full context including sensitive user data, API keys from responses, personally identifiable information. Logs become compliance liability.
Why it happens: Full context logging captures everything, including secrets.
Prevention: Context sanitization before storage. Remove credentials, PII, sensitive data. Log content hashes instead of full content for sensitive sources.
Summary & Next Steps
Context provenance logging solves the fundamental audit problem for agents: traditional logs capture what happened, but investigation requires understanding why. Actions alone don't explain intent for non-deterministic systems where decisions emerge from context.
The solution is logging context alongside actions. Every decision record includes references to context sources that influenced it. Context is stored separately with unique IDs. Investigation workflows can reconstruct full context windows and trace causation from context to decisions to actions.
The implementation requires context collectors that assign IDs to all context sources, decision loggers that record decisions with context provenance, context stores for full content, and query interfaces for investigation. The architecture separates audit records (metadata, references) from context storage (full content) to manage volume.
The operational challenges are log volume explosion requiring tiered storage, reference decay requiring synchronized retention, primary context misidentification requiring comprehensive logging, performance degradation requiring asynchronous writes, and privacy leakage requiring sanitization.
Here's what to build next:
Implement context ID assignment first: Before logging context, assign IDs. This is the foundation for provenance tracking.
Build decision-centric logging: Don't just log tool calls. Log decisions with context references, reasoning, and outcomes.
Create investigation tooling: Investigators need interfaces to query decisions by action, reconstruct context, trace context flow. Build these before incidents happen.
Establish retention policies: Decide how long to keep full context vs summaries vs metadata. Synchronize policies across audit DB and context store.
Monitor log volume: Track bytes logged per decision, total log volume, query latency. Optimize before volume becomes a problem.
Context provenance logging isn't optional for production agents—it's the only way to investigate incidents effectively. The question is whether you build it before or after discovering that action logs alone are insufficient for understanding agent behavior.
Related Articles
AI Security
- Capability Tokens: Fine-Grained Authorization for Non-Deterministic Agents
- Context Sandboxing: How to Prevent Tool Response Poisoning in Agentic Systems
- Credential Scoping for Agents: Why Temporary Keys Aren't Enough
- Prompt Injection Is Just the Beginning: The Undefendable Attack Surface of Agentic AI
Follow for more technical deep dives on AI/ML systems, production engineering, and building real-world applications: