SOC Agent — Journey Anomaly Detection & Automated Containment

Purple8 Hyper Graph ships a built-in Security Operations Centre (SOC) agent that monitors, classifies, and contains threats in real time across every active journey and tenant workload — with no external SIEM required and a mandatory human-release gate for all high-severity events.

Version

Introduced in v0.33.0 · Import: from purple8_graph.soc import SOCAgent

What the SOC Agent Does

The SOC agent is not a rules file you configure and forget. It is an active, self-calibrating security layer that runs on every metric observation and takes graduated action before an operator ever has to think about it:

Detects — Layer 1 rules catch threshold violations in under 1 ms. Layer 2 EWMA baselines catch subtle drift that rules miss.
Classifies — every signal is mapped to one of 10 canonical threat classes so the response is always proportionate.
Estimates blast radius — the agent calculates how many tenants, journeys, or SuperGraph peers could be affected before acting.
Contains — graduated actions from flagging (SOFT) to full tenant suspension and emergency snapshot (CRITICAL) are applied immediately.
Enforces human sign-off — HARD and CRITICAL containments cannot be auto-released. A named operator must call release().
Audits everything — every action, every classification, every release is written to an immutable AuditRecord log.

The agent is integrated into Hyper — it runs as part of the server; no separate wiring is required.

Architecture: How It Actually Works

┌─────────────────────────────────────────────────────────────────┐
│                    Journey / Tenant telemetry                    │
│  (failed_auth_count, query_result_size, cypher_query, …)        │
└──────────────────────────────┬──────────────────────────────────┘
                               │
              ┌────────────────▼─────────────────┐
              │   Layer 1 — Rule Engine           │  < 1 ms
              │   JourneyAnomalyRuleEngine        │  Stateless, SOC2-auditable
              │                                   │  11 threshold rules
              │   failed_auth_count ≥ 50  → HARD  │  Zero warm-up required
              │   DETACH DELETE in Cypher → CRIT  │
              └────────────────┬─────────────────┘
                               │ list[AnomalySignal]   (may be empty)
              ┌────────────────▼─────────────────┐
              │   Layer 2 — Statistical Baseline  │  Self-calibrating
              │   JourneyAnomalyDetector          │  Per-tenant, per-metric
              │                                   │  EWMA mean + variance
              │   Fires when z-score > 3.5 σ      │  30-sample warm-up guard
              └────────────────┬─────────────────┘
                               │ list[AnomalySignal]   (may be empty)
              ┌────────────────▼─────────────────┐
              │   ThreatClassifier                │
              │   Maps rule_name prefixes →       │
              │   (ThreatClass, ContainmentLevel) │
              │   Escalates to highest level seen │
              └────────────────┬─────────────────┘
                               │ (ThreatClass, ContainmentLevel)
              ┌────────────────▼─────────────────┐
              │   BlastRadiusEstimator            │
              │   Estimates affected tenants /    │
              │   journeys / SuperGraph peers     │
              └────────────────┬─────────────────┘
                               │ ThreatEvent (with blast_radius)
              ┌────────────────▼─────────────────┐
              │   ContainmentManager              │
              │   Executes graduated actions      │
              │   SOFT → MEDIUM → HARD → CRITICAL │
              │   Writes immutable AuditRecords   │
              │   Fires webhook / SIEM callbacks  │
              └────────────────┬─────────────────┘
                               │
              ┌────────────────▼─────────────────┐
              │   Human release gate              │
              │   HARD / CRITICAL: operator must  │
              │   call agent.release(event_id,    │
              │   released_by="name@org.com")     │
              └──────────────────────────────────┘

Key design principles:

Layer 1 always fires first — rule-based thresholds are explainable, auditable, and need no warm-up. This is the SOC2 compliance path.
Layer 2 adapts per tenant — EWMA baselines are per (tenant_id, metric_name). A tenant that legitimately runs large queries will not be flagged once its baseline settles.
Suspend first — every containment action suspends the journey or rate-limits the tenant before anything else. Humans must explicitly release HARD and CRITICAL events.
Full audit trail — every containment action and every release decision writes an AuditRecord. Nothing is silent.

Layer 1 — The Rule Engine

Layer 1 is deterministic. It runs on every process() call regardless of baseline state. Rules are pure functions — no memory, no I/O, sub-millisecond latency.

Each rule evaluates a key in the metrics dict, compares it to a threshold, and emits an AnomalySignal if the threshold is crossed.

The 11 Built-in Rules

Rule	Metric Key	Default Threshold	Threat Class
Credential stuffing	`failed_auth_count`	≥ 50 in 60 s	`credential_stuffing`
Result-size exfiltration	`query_result_size`	≥ 100 000 rows	`data_exfiltration`
Export-byte exfiltration	`export_bytes`	≥ 50 MB/min	`data_exfiltration`
RAG poisoning	`ingest_rate`	≥ 10 000 docs/min	`rag_poisoning`
Bulk delete	`delete_node_count`	≥ 500 nodes	`destructive_write`
Destructive Cypher	`cypher_query`	keyword match (`DETACH DELETE`, `DROP`, `TRUNCATE`, …)	`destructive_write`
Temporal insider	`admin_write_hour`	outside 06:00–22:00 UTC	`insider_threat`
Cross-label access	`distinct_labels_accessed`	≥ 20 distinct labels	`privilege_escalation`
Embedding inversion	`vector_query_rate`	≥ 300/min	`embedding_inversion`
Lateral movement	`peer_labels_queried` ∖ `peer_declared_domains`	any violation	`lateral_movement`
Insider write spread	`distinct_labels_written`	≥ 1 outside declared labels	`insider_threat`

Rules are SOC2 CC6.6 compliant — each AnomalySignal carries the rule_name, observed_value, threshold, and a UTC detected_at timestamp.

Why Rules Fire First

Rules fire before Layer 2 because they are immediately explainable. When a destructive_write signal reaches a CISO's inbox, the evidence is cypher_query = "DETACH DELETE n" — not a z-score. Layer 1 is the compliance path. Layer 2 fills in the behavioural gaps Layer 1 cannot cover.

Tune thresholds via RuleConfig:

python

from purple8_graph.soc import SOCAgent, SOCAgentConfig, RuleConfig

agent = SOCAgent(config=SOCAgentConfig(
    rule_config=RuleConfig(
        failed_auth_threshold=100,          # raise for high-traffic auth services
        bulk_delete_threshold=200,          # lower for stricter delete policies
    )
))

Layer 2 — The Statistical Detector

Layer 2 adapts to each tenant independently. It maintains a separate Exponentially Weighted Moving Average (EWMA) of mean and variance for every (tenant_id, metric_name) pair it observes.

The Math

$$\mu_t = \alpha \cdot x_t + (1-\alpha) \cdot \mu_{t-1}$$

$$\sigma^2_t = (1-\alpha) \cdot \left[\sigma^2_{t-1} + \alpha \cdot (x_t - \mu_{t-1})^2\right]$$

$$z = \frac{x_t - \mu_{t-1}}{\sqrt{\sigma^2_{t-1}}}$$

Critical implementation detail: the z-score is computed against μ_{t-1} before the current observation updates the baseline. A spike does not dilute its own detection signal.

Why This Matters

A tenant that runs 500 000-row exports as part of a normal batch pipeline would be flagged constantly by the Layer 1 rule (query_result_size ≥ 100 000). They can raise the Layer 1 threshold — or rely on Layer 2, which learns their baseline and fires only when their own normal is exceeded by 3.5σ. The same mechanism that protects small tenants from false positives also adapts to large tenants without weakening their protection.

Warm-up Guard

Layer 2 is silent for the first 30 samples per metric per tenant. During warm-up there is insufficient data to compute reliable variance. After warm-up the baseline is stable and z-score detection activates.

Tune via DetectorConfig:

python

from purple8_graph.soc import DetectorConfig

config = DetectorConfig(
    alpha=0.1,               # EWMA smoothing — lower = slower adaptation
    z_score_threshold=3.5,   # standard deviations before signal fires
    warmup_samples=30,       # samples required before Layer 2 activates
)

Threat Taxonomy — 10 Classes

`ThreatClass`	What It Means	Default Level	Blast Radius Logic
`credential_stuffing`	Brute-force authentication burst	HARD	1 + peers
`data_exfiltration`	Unusually large query result or export byte volume	HARD	all journeys for tenant
`privilege_escalation`	Actor accessing label set outside declared scope	MEDIUM	1
`lateral_movement`	SuperGraph peer querying labels outside declared domains	HARD	all tenants + peers
`insider_threat`	Admin write at unusual hours or across unexpected label set	MEDIUM	1
`rag_poisoning`	Ingest rate spike consistent with knowledge base poisoning	MEDIUM	1
`graph_dos`	Query or traversal rate anomaly consistent with resource exhaustion	MEDIUM	1
`tenant_boundary_violation`	Write or read crossing tenant isolation boundary	CRITICAL	all tenants + peers
`embedding_inversion`	Vector query rate anomaly consistent with embedding extraction probing	MEDIUM	1
`destructive_write`	Cypher contains `DETACH DELETE`, `DROP`, `TRUNCATE`, or similar	CRITICAL	all journeys for tenant

How Classification Works

ThreatClassifier scans each fired AnomalySignal's rule_name against a prefix table. When multiple signals fire in the same process() call:

Threat class: first match wins (most specific rule)
Containment level: maximum across all signals — always escalates, never degrades

A combination of a MEDIUM privilege_escalation signal and a HARD data_exfiltration signal in the same observation produces a HARD containment — not an average.

Containment — Graduated Response

The Four Levels

Level	`requires_human_release`	What Happens
SOFT	No	Journey flagged in audit log. Alert emitted. No service disruption.
MEDIUM	No	Rate limit applied to tenant. Journey continues under throttle. Alert emitted.
HARD	Yes	Writes suspended. Peer connection isolated. Operator must release.
CRITICAL	Yes	Tenant fully suspended. API tokens revoked. Emergency snapshot triggered. Multi-team escalation. Operator must release.

Actions Available

Action	Triggered at
`FLAG_JOURNEY`	SOFT
`EMIT_ALERT`	SOFT+
`INCREMENT_THREAT_SCORE`	SOFT+
`RATE_LIMIT_TENANT`	MEDIUM
`QUARANTINE_LABEL`	MEDIUM
`SUSPEND_WRITES`	HARD
`ISOLATE_PEER`	HARD (lateral movement)
`SUSPEND_TENANT`	CRITICAL
`REVOKE_API_TOKENS`	CRITICAL
`EMERGENCY_SNAPSHOT`	CRITICAL (when `auto_snapshot=True`)

The Audit Trail

Every action — including every release — writes an immutable AuditRecord:

python

@dataclass
class AuditRecord:
    record_id:    str               # UUID
    event_id:     str               # → ThreatEvent
    action:       ContainmentAction
    level:        ContainmentLevel
    threat_class: ThreatClass
    tenant_id:    str
    journey_id:   str
    actor:        str               # "soc_agent" or operator name
    released_by:  str | None        # None until human releases
    released_at:  datetime | None
    notes:        str
    timestamp:    datetime          # UTC

The log never shrinks. released_at is appended on release; the original containment record is never modified.

Blast Radius Estimation

Before applying any containment, BlastRadiusEstimator calculates how many resources are at risk:

Threat Class	Blast Radius Formula
`tenant_boundary_violation`, `lateral_movement`	`known_tenant_count + known_peer_count`
`data_exfiltration`, `destructive_write`	`known_journey_count` (all journeys for tenant)
`credential_stuffing`	`1 + known_peer_count`
All others	`1` (journey-scoped)

Configure for your topology:

python

from purple8_graph.soc import BlastRadiusEstimator, SOCAgentConfig, SOCAgent

estimator = BlastRadiusEstimator(
    known_tenant_count=100,
    known_journey_count=500,
    known_peer_count=4,       # SuperGraph peers
)
agent = SOCAgent(config=SOCAgentConfig(blast_radius=estimator))

Human-Release Gate — How It Works

HARD and CRITICAL events cannot be released by the system. ContainmentLevel.requires_human_release is True for both levels, and release() enforces a non-empty released_by string — anonymous releases raise ValueError by design.

python

# See what is currently locked
for state in agent.active_containments():
    print(f"{state.event_id}  level={state.level}  tenant={state.tenant_id}")

# Release after human review and root-cause confirmation
records = agent.release(
    event_id="evt-abc123",
    released_by="soc-analyst@acme.com",
    notes="Source IP confirmed malicious. WAF rule deployed. No data exfiltrated.",
)
# Every AuditRecord for this event now has .released_by and .released_at populated.

Quick Start

Basic usage

python

from purple8_graph.soc import SOCAgent

agent = SOCAgent()

# Feed metrics from a running journey
result = agent.process(
    metrics={"failed_auth_count": 300},
    tenant_id="acme",
    journey_id="j-001",
)

print(result["threat_class"])       # "credential_stuffing"
print(result["containment_level"])  # "hard"
print(result["contained"])          # True

# HARD / CRITICAL events must be explicitly released by a human
agent.release(
    event_id=result["event_id"],
    released_by="admin@acme.com",
    notes="Confirmed brute-force. Source IP blocked at WAF.",
)

Full enterprise configuration

python

from purple8_graph.soc import (
    SOCAgent, SOCAgentConfig,
    RuleConfig, DetectorConfig, ContainmentConfig,
    BlastRadiusEstimator,
)

estimator = BlastRadiusEstimator(
    known_tenant_count=50,
    known_journey_count=200,
    known_peer_count=3,
)

agent = SOCAgent(config=SOCAgentConfig(
    rule_config=RuleConfig(
        failed_auth_threshold=100,
        query_result_size_threshold=500_000,
    ),
    detector_config=DetectorConfig(
        alpha=0.05,           # slower adaptation for stable tenants
        z_score_threshold=4.0,
        warmup_samples=50,
    ),
    containment_config=ContainmentConfig(
        webhook_url="https://hooks.acme.com/soc-alerts",
        siem_callback=my_siem_push,   # callable(AuditRecord)
        auto_snapshot=True,
    ),
    blast_radius=estimator,
    enable_layer1=True,
    enable_layer2=True,
))

result = agent.process(
    metrics={
        "bytes_transferred": 15 * 1024 ** 3,   # 15 GiB — triggers data_exfiltration
        "vector_query_rate": 2500,
    },
    tenant_id="enterprise-corp",
    journey_id="rag-pipeline-7",
    peer_instance_ids=["peer-eu-1", "peer-ap-1"],
)

Incident report for SIEM / ticketing

python

report = agent.incident_report(result["event_id"])
# {
#   "event_id": "...",
#   "threat_class": "data_exfiltration",
#   "containment_level": "hard",
#   "blast_radius": 4,
#   "tenant_id": "enterprise-corp",
#   "journey_id": "rag-pipeline-7",
#   "signals": [...],
#   "audit_records": [...],
#   "is_contained": True,
#   "is_released": False,
# }

Human-Release Gate

Any HARD or CRITICAL event cannot be auto-released. ContainmentLevel.requires_human_release is True for both levels. The release path is:

python

# Check what is currently contained
for state in agent.active_containments():
    print(state.event_id, state.level, state.applied_at)

# Release a specific event
records = agent.release(
    event_id="evt-abc123",
    released_by="soc-analyst@acme.com",
    notes="Root cause confirmed; firewall rule deployed.",
)
# Each AuditRecord now has .released_by and .released_at set

Calling release() with an empty released_by string raises ValueError — anonymous releases are rejected by design.

SOC2 Control Mapping

SOC2 Control	Purple8 Mechanism
CC6.1 — Logical access controls	`REVOKE_API_TOKENS`, `SUSPEND_TENANT` actions
CC6.6 — Anomaly detection	Layer 1 rule engine (11 rules), Layer 2 EWMA z-score
CC6.7 — Privileged access	`SUSPEND_WRITES` gate; human release required for HARD/CRITICAL
CC6.8 — Malware / destructive writes	`destructive_write` rule, `QUARANTINE_LABEL`, `EMERGENCY_SNAPSHOT`
CC7.2 — System monitoring	`AuditRecord` log, webhook + SIEM callbacks
CC7.3 — Incident response	`incident_report()`, graduated containment ladder
CC7.4 — Recovery	`EMERGENCY_SNAPSHOT` on CRITICAL; operator release with audit evidence
CC9.2 — Business continuity	Emergency snapshot before any CRITICAL suspension

The Dashboard API

The SOC router (routers/soc.py) exposes five REST endpoints used by the Hyper SOC Dashboard:

Endpoint	What It Returns
`GET /soc/containments/active`	All un-released containments — live threat feed
`GET /soc/audit-log`	Full immutable audit log with pagination
`GET /soc/incident/{event_id}`	Drill-down report for one event — signals, actions, release status
`GET /soc/stats`	Aggregate KPI counts by threat class and level
`POST /soc/containments/{event_id}/release`	Human release gate — requires `released_by` in body

All responses use camelCase aliases so the React frontend consumes them without transformation.

Observability

The SOC module exposes structured data for your own OTEL/Prometheus pipeline:

python

# Full immutable audit log — never shrinks
for record in agent.audit_log:
    print(record.record_id, record.action, record.threat_class, record.timestamp)

# Currently active (un-released) containments
for state in agent.active_containments():
    print(state.containment_id, state.level, state.tenant_id, state.applied_at)

# Check if a specific event is still contained
agent.is_contained("evt-abc123")   # True / False

Configuration Reference

`RuleConfig` — Layer 1 Thresholds

Parameter	Default	Description
`failed_auth_window_secs`	`60`	Window for counting failed auth attempts
`failed_auth_threshold`	`50`	Max failed auth attempts before HARD alert
`query_result_size_threshold`	`100_000`	Max rows per query
`ingest_rate_per_min`	`10_000`	Max docs ingested per minute per tenant
`bulk_delete_threshold`	`500`	Min node count to trigger destructive-write rule
`exfil_bytes_per_min`	`50_000_000`	Max bytes exported per minute (50 MB)
`cross_label_access_threshold`	`20`	Max distinct labels per actor per minute
`vector_query_rate_per_min`	`300`	Max vector queries per minute per actor
`stage_duration_multiplier`	`10.0`	SLA multiplier before temporal anomaly fires
`destructive_pattern_threshold`	`1`	Any destructive keyword match fires

`DetectorConfig` — Layer 2 EWMA

Parameter	Default	Description
`alpha`	`0.1`	EWMA smoothing factor — lower = slower adaptation
`z_score_threshold`	`3.5`	Standard deviations before a signal fires
`warmup_samples`	`30`	Observations required before Layer 2 activates

`ContainmentConfig` — Notifications

Parameter	Default	Description
`webhook_url`	`None`	POST incident JSON here on every action
`siem_callback`	`None`	`callable(AuditRecord)` for SIEM integration
`auto_snapshot`	`True`	Trigger emergency SST snapshot on CRITICAL
`notify_on_release`	`True`	Fire webhook/SIEM when a containment is released

API Reference

`SOCAgent`

Method	Returns	Description
`process(metrics, tenant_id, journey_id, peer_instance_ids)`	`dict`	Run full pipeline; returns signals, classification, audit records
`release(event_id, released_by, notes)`	`list[AuditRecord]`	Human-release gate for HARD/CRITICAL events
`incident_report(event_id)`	`dict`	Full structured report for SIEM / ticketing
`audit_log`	`list[AuditRecord]`	Immutable log of all containment actions
`active_containments()`	`list[ContainmentState]`	Un-released containment states
`is_contained(event_id)`	`bool`	Check if event still has active containment

`process()` return dict

Key	Type	Description
`signals`	`list[AnomalySignal]`	All fired signals (Layer 1 + Layer 2)
`threat_class`	`str`	Classified threat class name
`containment_level`	`str`	`"soft"` / `"medium"` / `"hard"` / `"critical"`
`event_id`	`str \| None`	UUID of the `ThreatEvent`; `None` if no signals fired
`blast_radius`	`int`	Estimated affected resource count
`audit_records`	`list[AuditRecord]`	Records written during this `process()` call
`contained`	`bool`	Whether containment actions were applied

Module Map

Module	Key Classes	Purpose
`soc/threats.py`	`ThreatClass`, `ContainmentLevel`, `ContainmentAction`, `AnomalySignal`, `ThreatEvent`, `AuditRecord`	Shared types — foundation for all other modules
`soc/rules.py`	`RuleConfig`, `JourneyAnomalyRuleEngine`	Layer 1 — stateless threshold rules
`soc/detector.py`	`DetectorConfig`, `MetricBaseline`, `JourneyAnomalyDetector`	Layer 2 — EWMA per-tenant baselines
`soc/containment.py`	`ContainmentConfig`, `ContainmentState`, `ContainmentManager`	Graduated actions, audit log, human release gate
`soc/agent.py`	`BlastRadiusEstimator`, `ThreatClassifier`, `SOCAgentConfig`, `SOCAgent`	Orchestrator — wires all layers into `process()`
`soc/router.py`	FastAPI router	REST endpoints for the SOC Dashboard

← Journey Engine | SuperGraph Federation →

SOC Agent — Journey Anomaly Detection & Automated Containment ​

What the SOC Agent Does ​

Architecture: How It Actually Works ​

Layer 1 — The Rule Engine ​

The 11 Built-in Rules ​

Why Rules Fire First ​

Layer 2 — The Statistical Detector ​

The Math ​

Why This Matters ​

Warm-up Guard ​

Threat Taxonomy — 10 Classes ​

How Classification Works ​

Containment — Graduated Response ​

The Four Levels ​

Actions Available ​

The Audit Trail ​

Blast Radius Estimation ​

Human-Release Gate — How It Works ​

Quick Start ​

Basic usage ​

Full enterprise configuration ​

Incident report for SIEM / ticketing ​

Human-Release Gate ​

SOC2 Control Mapping ​

The Dashboard API ​

Observability ​

Configuration Reference ​

RuleConfig — Layer 1 Thresholds ​

DetectorConfig — Layer 2 EWMA ​

ContainmentConfig — Notifications ​

API Reference ​

SOCAgent ​

process() return dict ​

Module Map ​

SOC Agent — Journey Anomaly Detection & Automated Containment

What the SOC Agent Does

Architecture: How It Actually Works

Layer 1 — The Rule Engine

The 11 Built-in Rules

Why Rules Fire First

Layer 2 — The Statistical Detector

The Math

Why This Matters

Warm-up Guard

Threat Taxonomy — 10 Classes

How Classification Works

Containment — Graduated Response

The Four Levels

Actions Available

The Audit Trail

Blast Radius Estimation

Human-Release Gate — How It Works

Quick Start

Basic usage

Full enterprise configuration

Incident report for SIEM / ticketing

Human-Release Gate

SOC2 Control Mapping

The Dashboard API

Observability

Configuration Reference

`RuleConfig` — Layer 1 Thresholds

`DetectorConfig` — Layer 2 EWMA

`ContainmentConfig` — Notifications

API Reference

`SOCAgent`

`process()` return dict

Module Map