How AgentObserver Orchestrates Multiple AI Agent Parsers in ADR

The AgentObserver class in Uber's ADR repository provides a centralized orchestration layer that instantiates multiple agent-specific parsers, executes them selectively, and aggregates their outputs into a unified stream of AgentEvent objects.

The ADR (AI Development Records) Sensor is designed to capture and analyze interactions from diverse AI coding assistants. At the heart of this system sits AgentObserver—a single coordinator that abstracts away the complexity of disparate log formats from Claude, Claude Desktop, Cursor, Cline, Warp, and Codex. This article examines how this orchestration works, drawing directly from the source code at uber/ADR.

Parser Instantiation in the Constructor

When AgentObserver is initialized, it eagerly creates parser instances for every supported agent. This design ensures all dependencies are ready before any ingestion begins.

In observer.py lines 50–57, the __init__ method constructs parsers with optional age-based filtering:

self.claude_parser = ClaudeParser(max_age_days=max_age_days) if max_age_days else ClaudeParser()
self.cursor_parser = CursorParser(max_age_days=max_age_days) if max_age_days else CursorParser()
self.claude_desktop_parser = ClaudeDesktopParser(max_age_days=max_age_days) if max_age_days else ClaudeDesktopParser()
self.codex_parser = CodexParser()
self.cline_parser = ClineParser()
self.warp_parser = WarpParser(max_age_days=max_age_days) if max_age_days else WarpParser()

Not all parsers support age filtering—CodexParser and ClineParser are instantiated without this parameter, reflecting their simpler log management models.

Controlled Ingestion via ingest_all

The ingest_all method serves as the primary entry point for AgentObserver orchestration. It accepts a source_filter parameter (default "all") that enables selective parser execution.

The method follows a consistent pattern for each agent, as seen in observer.py lines 88–115 and 124–151:

if source_filter in ["all", "claude"]:
    claude_entries = self.claude_parser.parse_all()

For each parser invocation, the orchestration pipeline performs three critical operations:

• Execution – Calls parse_all() on the specific parser
• Validation – Filters entries through has_meaningful_content() to discard empty conversations
• Error Isolation – Wraps each block in try/except with _emit_error logging to prevent one failing parser from crashing the entire pipeline

This defensive design ensures robustness across heterogeneous log sources with varying quality and availability.

Result Aggregation and Unified Export

After all enabled parsers complete, ingest_all returns a standardized tuple:

return all_entries, system_config_data

Both elements derive from the same abstraction: every parser emits AgentEvent objects defined in schemas/agent_event_schema.py. This unified type system allows downstream consumers—whether analytics pipelines, dashboards, or archival tools—to process Claude logs and Cursor logs identically.

The calling code then has access to three export methods:

• display_summary(events, configs) – Formatted console output
• save_to_file(events, configs, output_format) – Consolidated file export (supports JSONL)
• save_sessions_to_individual_files(events) – Per-session file generation

Key Orchestration Mechanisms

Mechanism	Purpose	Location
Parser objects	Encapsulate agent-specific log parsing logic	__init__ method
source_filter	Enable targeted ingestion (e.g., "warp" only)	ingest_all signature
Content filtering	Remove noise via has_meaningful_content()	Post-parsing validation
Error handling	_emit_error writes to error.log without raising	Per-parser try/except blocks
AgentEvent schema	Unified data model across all agents	schemas/agent_event_schema.py

Practical Usage Examples

Basic ingestion with summary display:

from adr_sensor.observer import AgentObserver

observer = AgentObserver()                 # default output directory: "output"

events, configs = observer.ingest_all()    # ingest all supported agents

observer.display_summary(events, configs)  # formatted results table

Targeted single-agent extraction with JSON Lines export:

observer = AgentObserver()
warp_events, _ = observer.ingest_all(source_filter="warp")
saved = observer.save_to_file(warp_events, [], output_format="jsonl")
print(f"Saved {len(saved)} files")

Extensibility Architecture

Adding support for a new AI coding agent requires only two steps:

1. Implement a parser class following the BaseParser contract
2. Register it in AgentObserver.__init__ and ingest_all

No changes are needed to downstream consumers because the AgentEvent abstraction shields them from parser-specific details. This plug-and-play design has enabled ADR to expand from initial Claude support to six distinct agents without architectural changes.

File Structure Reference

File	Responsibility
Sensor/adr_sensor/observer.py	Central orchestrator
Sensor/adr_sensor/parsers/claude_parser.py	Claude Code JSONL parsing
Sensor/adr_sensor/parsers/cursor_parser.py	Cursor log parsing
Sensor/adr_sensor/parsers/claude_desktop_parser.py	Claude Desktop (macOS) logs
Sensor/adr_sensor/parsers/cline_parser.py	Cline agent parsing
Sensor/adr_sensor/parsers/warp_parser.py	Warp Terminal logs
Sensor/adr_sensor/parsers/codex_parser.py	OpenAI Codex logs
Sensor/adr_sensor/schemas/agent_event_schema.py	Unified event model

Summary

• AgentObserver instantiates parsers lazily in __init__, then coordinates them through ingest_all
• Selective execution via source_filter enables efficient targeted analysis without code changes
• Defensive error handling isolates parser failures via _emit_error logging
• Unified AgentEvent schema eliminates format heterogeneity for downstream consumers
• Plug-and-play architecture requires only parser implementation and registration to add new agents

Frequently Asked Questions

What happens if one parser fails during ingest_all?

The failure is caught, logged via _emit_error to error.log as a one-line JSON record, and processing continues with remaining parsers. This ensures a single corrupted log file doesn't halt the entire ingestion pipeline.

How does AgentObserver handle agents with different configuration options?

The constructor applies max_age_days only to parsers that support it, using conditional instantiation. Parsers without age filtering—like CodexParser and ClineParser—receive simpler initialization. Source-specific logic in ingest_all accommodates varying method signatures.

Can I ingest from multiple specific agents without processing all of them?

The source_filter parameter accepts a single string value. To target multiple specific agents, make separate ingest_all calls with different filters and aggregate the results manually, or use "all" and filter the returned events by agent type.

Where is the unified data model defined?

All parsers emit AgentEvent objects as defined in Sensor/adr_sensor/schemas/agent_event_schema.py. This schema standardizes fields like timestamp, agent type, conversation content, and metadata across every supported AI assistant.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}