Streaming pattern detection for AI agent actions across systems.
Catches multi-step risks that single-action authorizers cannot see. Each individual tool call might pass any policy check in isolation. The correlator looks at sequences over time and surfaces patterns: this then that, within X minutes, by the same agent.
Agent actions flow in:
─────────────────────────────────────────────────────►
notion.read_sensitive(page_secret) ◄── normal
gmail.send(lawyer@external.com) ◄── normal
─────────────────────────────────────────────────────►
⚠️ EXFILTRATION
read sensitive → send external
same agent, within 25s
Agent tool calls are usually authorized one at a time. A read is a read; a send is a send. Both can be perfectly fine in isolation.
But many real agent risks are sequences:
- Exfiltration — read sensitive data, then send to an external address
- Privilege escalation — check what permissions exist, then immediately use the destructive one
- Credential abuse — read a secret, then make an outbound network call
- Runaway loops — N destructive actions in 30 seconds
Single-action authorizers (allow/deny per tool call, including the Wave engine and most enterprise governance products) cannot see these because each individual step looks fine. You need a layer that watches the stream.
This library is that layer. Small, deterministic, no dependencies, drop-in.
pip install cross-system-correlatorPython 3.9+. Zero runtime dependencies.
from correlator import (
Correlator, AgentAction, exfiltration_pattern,
)
# Build the correlator with one or more patterns
correlator = Correlator(patterns=[exfiltration_pattern()])
# Register an alert handler
correlator.on_match(lambda m: print(f"ALERT [{m.severity.value}]: {m.pattern_name}"))
# Pipe every agent action through observe()
correlator.observe(AgentAction(
agent_id="research-agent",
system="notion",
action="read_page",
action_class="read_sensitive",
target="page_12345",
))
correlator.observe(AgentAction(
agent_id="research-agent",
system="gmail",
action="send",
target="lawyer@external.com",
metadata={"external": True},
))
# → ALERT [high]: possible_exfiltrationThe correlator is system-agnostic. Feed it any event stream where each event has at minimum: agent identity, system, action name, and a timestamp. Optional but useful: action_class (high-level category) and target (resource id, recipient, etc).
AgentAction(
agent_id="research-agent",
system="notion",
action="read_page",
action_class="read_sensitive", # optional category
target="page_12345", # optional target
metadata={"external": True}, # arbitrary extra fields
timestamp=time.time(), # defaults to now
)The action_class field is how patterns can be portable across systems. A pattern matching read_sensitive will fire whether the sensitive read came from Notion, S3, Postgres, or anywhere else you've tagged.
The library ships with six patterns ready to use:
| Pattern | Severity | Catches |
|---|---|---|
exfiltration_pattern() |
HIGH | sensitive read → external send (same agent) |
data_egress_pattern() |
HIGH | sensitive read → cloud-storage write (same agent) |
credential_then_external_pattern() |
HIGH | credential read → outbound network call |
cross_agent_data_handoff_pattern() |
HIGH | sensitive read by one agent → external send by any agent |
privilege_escalation_pattern() |
MEDIUM | permission check → destructive action |
rapid_destructive_burst_pattern() |
CRITICAL | 3 destructive actions in 30 seconds |
Drop in whichever fit your environment:
from correlator import (
Correlator,
exfiltration_pattern,
privilege_escalation_pattern,
rapid_destructive_burst_pattern,
)
correlator = Correlator(patterns=[
exfiltration_pattern(window_seconds=120),
privilege_escalation_pattern(window_seconds=60),
rapid_destructive_burst_pattern(),
])Each pattern has tunable parameters (window size, etc).
A Pattern is an ordered list of PatternSteps plus a time window:
from correlator import (
Pattern, PatternStep, Severity,
match_system, match_action, match_all, match_target_equals_context,
capture_target,
)
share_after_read = Pattern(
name="document_shared_after_read_by_same_agent",
description="Same agent reads then shares the same document.",
severity=Severity.MEDIUM,
window_seconds=300,
agent_scope="same", # both steps must come from same agent_id
steps=[
PatternStep(
name="read_doc",
matcher=match_all(match_system("notion"), match_action("read")),
capture=capture_target("doc"), # remember which doc
),
PatternStep(
name="share_same_doc",
matcher=match_all(
match_system("notion"),
match_action("share"),
match_target_equals_context("doc"), # must match captured doc
),
),
],
)The pattern engine handles ordering, agent scoping, time windows, and context passing between steps. You just declare what to look for.
These compose into step matchers without writing lambdas:
match_system("notion", "s3") # glob patterns
match_action("read*", "delete_*")
match_action_class("read_sensitive")
match_target_prefix("@external")
match_target_regex(r"prod-.*-bucket")
match_target_equals_context("doc") # equals a capture from earlier step
match_metadata_equals("external", True)
match_metadata("amount", lambda v: v > 1000)
match_all(match_system("aws"), match_action_class("destructive"))
match_any(match_system("gmail"), match_system("slack"))A matcher is just Callable[[AgentAction, dict], bool]. Write your own if needed.
git clone https://github.com/andreas-altamirano/cross-system-correlator
cd cross-system-correlator
pip install -e .
python examples/simulate.pyThe demo replays 12 agent actions through the correlator. Most are benign. Three planted attack sequences (exfiltration, privilege escalation, rapid destructive burst) fire alerts in real time as they complete. You'll see output like:
══════════════════════════════════════════════════════════════════════
ALERT [HIGH] possible_exfiltration
Sensitive read followed by external send by the same agent.
Agent: research-agent
Duration: 25.0s across 2 actions
1. notion.read_page [read_sensitive] → page_secret_strategy
2. gmail.send → lawyer@external.com
══════════════════════════════════════════════════════════════════════
The engine maintains a small set of pending partial matches. When an action arrives:
- Advance pending matches. For each in-progress partial, check whether this action satisfies the next required step. If yes, advance; if it completes the pattern, fire. If the partial has aged past the window, drop it.
- Seed new pending matches. For each registered pattern, check whether this action matches the first step. If yes, start a new pending match (locked to this agent if
agent_scope="same"). - Fire handlers for any completed matches.
Memory is O(active partials), bounded by max_partial_matches. CPU is O(patterns + active partials) per action. The whole thing is in-process, lock-protected for thread safety, and has zero dependencies.
For high-throughput environments (>10k actions/sec) you'd want to shard across multiple correlator instances by agent_id or partition the pattern set. The default settings are fine for the kind of action volumes a real-world agent fleet produces.
The correlator is independent — feed it actions however you like. Common patterns:
Sidecar log tailer. Have your agent log every tool call as JSON to stdout / a file / a queue. A small tailer process parses and feeds them to correlator.observe(). Alerts go to Slack, PagerDuty, your SIEM, etc.
Inline with mcp-governance-proxy. If you're already using mcp-governance-proxy to evaluate tool calls, add a hook that also calls correlator.observe() on every action. You get single-action authorization (proxy) + multi-step correlation (this library) in one place.
Pipe from wave-engine decisions. If you've already integrated wave-engine, every Decision you produce is implicitly an action. Construct an AgentAction from it and observe.
- Not a tool authorizer. It observes and alerts; it does not block actions in flight. Combine with a pre-action authorizer if you need to refuse calls. (See wave-engine and mcp-governance-proxy.)
- Not a SIEM. It's a focused pattern engine for agent actions. Pipe alerts to your existing SIEM/notification stack.
- Not a persistence layer. Partial matches live in-process. If the process restarts, in-flight partials are lost. For most use cases this is fine — patterns fire within seconds-to-minutes of the first step.
- Not a content classifier. It doesn't decide which Notion pages are "sensitive." You tag them with
action_class="read_sensitive"upstream; this library trusts that label.
pip install -e ".[dev]"
pytest -q21 tests covering single-step matches, multi-step ordering, window expiration, agent scoping, context captures, handler callbacks, error resilience, and all six pre-built patterns.
Apache 2.0. See LICENSE.
Extracted and generalized from the cross-system correlation engine inside Surfit, an AI agent governance platform. Released so other teams can detect multi-step agent risks without building this from scratch.
Companion libraries:
wave-engine— continuous 1–5 risk scoring per actionmcp-governance-proxy— MCP server that proxies tool calls through a pluggable evaluator