How to Filter ADR Agent Events by Source (Claude, Cursor, Codex)

Filter Uber ADR AgentEvent objects by checking the source attribute against agent identifiers like "claude", "cursor", or "codex" using standard Python list comprehensions.

The Uber ADR Sensor library captures telemetry from AI coding agents and represents each record as an AgentEvent dataclass. Since the source field stores the originating agent identifier, you can filter ADR agent events by source with simple Python collection operations—no complex query language required.

---

Understanding the source Field in AgentEvent

The AgentEvent dataclass in Sensor/adr_sensor/schemas/agent_event_schema.py defines a source attribute at line 49 that stores the agent identifier as a plain string:

# From adr_sensor/schemas/agent_event_schema.py (line 49)

@dataclass(frozen=True)
class AgentEvent:
    source: str           # "claude", "cursor", "codex", etc.

    # ... other fields

Because @dataclass(frozen=True) makes the schema immutable, the source value is set once during event creation and cannot be modified afterward. This guarantees consistent, deterministic filtering.

Individual parsers populate this field:

• Claude Parser (claude_parser.py) sets source="claude"
• Cursor Parser (cursor_parser.py) sets source="cursor"
• Codex Parser (codex_parser.py) sets source="codex"

---

Filtering Events After Ingestion

The AgentObserver API provides ingest_all() to load events from log directories. Apply a list comprehension to filter by source:

from adr_sensor import AgentObserver

# Load all events from the default log directory

observer = AgentObserver()
all_events, _ = observer.ingest_all()  # Returns (list[AgentEvent], list[SystemConfig])

# Filter for Claude, Cursor, or Codex events

allowed_sources = {"claude", "cursor", "codex"}
filtered_events = [e for e in all_events if e.source in allowed_sources]

print(f"Total events: {len(all_events)}")
print(f"Filtered events: {len(filtered_events)}")

# Display summary of filtered results

observer.display_summary(filtered_events, [])

This approach works because AgentEvent objects are plain Python dataclasses with public attributes.

---

Reusable Filtering Function

For scripts and notebooks, extract the logic into a reusable function:

def filter_events_by_source(events, sources):
    """
    Return AgentEvent objects whose source matches any entry in sources.
    
    Parameters:
        events: list[AgentEvent] - events to filter
        sources: list[str] or set[str] - agent identifiers to keep
    
    Returns:
        list[AgentEvent] - filtered events
    """
    source_set = set(sources)  # O(1) lookup

    return [e for e in events if e.source in source_set]


# Usage example

from adr_sensor import AgentObserver

observer = AgentObserver()
events, _ = observer.ingest_all()

# Filter for specific agents

target_agents = ["claude", "cursor", "codex"]
filtered = filter_events_by_source(events, target_agents)

# Pass to downstream analysis or export

observer.display_summary(filtered, [])

Using set(sources) ensures O(1) membership testing for large event volumes.

---

Filtering Custom Event Collections

If you generate events outside the standard ingestion pipeline—for example, using the demo utilities in Sensor/examples/demo.py—the same pattern applies:

from examples.demo import create_sample_events

# Create synthetic events with various sources

events = create_sample_events()

# Select only Claude, Cursor, and Codex

agent_events = [e for e in events if e.source in {"claude", "cursor", "codex"}]

This flexibility lets you integrate source filtering into custom parsers, testing suites, or data pipelines.

---

Key Source Files

File	Purpose
Sensor/adr_sensor/schemas/agent_event_schema.py	Defines AgentEvent dataclass with source field (line 49)
Sensor/adr_sensor/parsers/claude_parser.py	Sets source="claude" during event construction
Sensor/adr_sensor/parsers/cursor_parser.py	Sets source="cursor" during event construction
Sensor/adr_sensor/parsers/codex_parser.py	Sets source="codex" during event construction
Sensor/examples/demo.py	Provides create_sample_events() for testing filters

---

Performance Considerations

• Time complexity: O(n) where n = number of events; set lookup for source matching is O(1)
• Memory: List comprehensions create new lists; for large datasets, consider generator expressions: (e for e in events if e.source in allowed)
• Frozen dataclasses: Immutable AgentEvent objects enable safe sharing across threads without copying

---

Summary

• The source attribute in AgentEvent stores agent identifiers as strings: "claude", "cursor", "codex"
• Filter with standard Python: [e for e in events if e.source in {"claude", "cursor", "codex"}]
• AgentObserver.ingest_all() returns events ready for source-based filtering
• Extract reusable functions for scripts, notebooks, and automated pipelines
• Immutable schema guarantees consistent, thread-safe filtering

---

Frequently Asked Questions

How does the ADR Sensor know which agent created an event?

Each parser hardcodes its agent identifier. The Claude Parser sets source="claude", the Cursor Parser sets source="cursor", and the Codex Parser sets source="codex" when constructing AgentEvent objects. This design ensures consistent identifiers across the codebase regardless of log format variations.

Can I filter events before ingestion instead of after?

The AgentObserver.ingest_all() method loads all events first. For pre-ingestion filtering, you would need to implement custom logic at the parser or file-scanning level—ADR does not currently expose built-in source filters during ingestion. Post-ingestion filtering with list comprehensions is the recommended approach.

What happens if I pass an unknown source name to my filter?

The filter simply excludes events with non-matching sources. No error is raised. If you need to validate source names against known agents, inspect the parser files or maintain your own allowlist: {"claude", "cursor", "codex", "copilot", "gpt-engineer", ...}.

Are AgentEvent objects modifiable after creation?

No. The @dataclass(frozen=True) decorator prevents field modification after instantiation. This immutability makes source filtering safe for concurrent processing and ensures audit trail integrity in security-sensitive deployments.

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