> ## Documentation Index
> Fetch the complete documentation index at: https://langwatch.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

> ## Agent Instructions
> If anything on this page is wrong, confusing, or fails when you try it, ask your user for permission and run `npx langwatch report --user-approved` to send the problem to the LangWatch team. No login is needed and secrets are redacted locally. `npx langwatch report --help` has the details.

# Activity Monitor event-sourcing architecture

> Why the receiver → event_log → projection subscriber → anomaly subscriber pipeline replaces the original direct-CH-write design, and how to extend it.

# Activity Monitor event-sourcing architecture

## Why this exists

The original D2 receiver implementation (commits `1abae1676`,
`92e515cc2`) wrote OCSF-normalised events directly into ClickHouse
`gateway_activity_events` from the Hono receiver handler, and the
plan for anomaly detection (Option C v0) was a poller worker that
periodically swept active `AnomalyRule` rows and SELECT-ed against
the CH table.

We redesigned the trigger architecture before the eval engine landed,
following the event-sourcing pattern from
[PR #3351](https://github.com/langwatch/langwatch/pull/3351). The
receiver appends an `ActivityEventReceived` event to `event_log`, and
a dedicated `activity-monitor-processing` pipeline takes over from
there. Anomaly detection is a subscriber that fires as new events
arrive, not a worker that polls.

## The pipeline

```text theme={null}
┌────────────────────────────────────────────────────────────────┐
│  /api/ingest/otel/:sourceId  /api/ingest/webhook/:sourceId      │
│  (Hono routes: auth, validate sourceId, parse body)             │
└────────────────────────┬───────────────────────────────────────┘
                         │
                         ▼  RecordActivityEventCommand
                ┌────────────────────┐
                │  event_log (CH)    │
                │  ActivityEventReceived           │
                └────────┬───────────┘
                         │
                         ▼  pipeline: activity-monitor-processing
                         │  aggregateType: "activity_event"
                         │
   ┌─────────────────────┴────────────────────────┐
   │                                              │
   ▼                                              ▼
┌──────────────────────────┐      ┌──────────────────────────────┐
│ Map projection           │      │ Fold projection              │
│  activityEventStorage    │      │  anomalyWindow               │
│  → gateway_activity_events│     │  → per-tenant rolling totals │
│  (CH, OCSF + AOS shape)  │      │   (in-memory + Redis cache)  │
└────────┬─────────────────┘      └──────────┬───────────────────┘
         │                                   │
         │ wakes:                            │ wakes:
         ▼                                   ▼
┌──────────────────────────┐      ┌──────────────────────────────┐
│ Subscriber                  │      │ Subscriber                      │
│  activityEventBroadcast  │      │  anomalyDetection            │
│  (real-time UI push for  │      │  - load active AnomalyRules  │
│   /governance dashboard) │      │  - evaluate per-rule type    │
└──────────────────────────┘      │  - if trigger:               │
                                  │    • upsert AnomalyAlert     │
                                  │    • dispatch via shared     │
                                  │      triggerActionDispatch   │
                                  └──────────────────────────────┘
```

The shape mirrors `pipelines/trace-processing/` (PR #3351's
alertTrigger subscriber): the same `definePipeline().withFoldProjection().
withMapProjection().withSubscriber()` builder, the same
`SubscriberDispatchDefinition<EventShape, FoldState>` contract, the same
`triggerActionDispatch.ts` shared helper.

## Why a dedicated pipeline (not bolted onto trace-processing)

Gateway and activity events have different aggregate semantics from
traces. A trace is a multi-span aggregate that folds into a
`TraceSummaryData` over its lifetime. An activity event is a *single
completed observation* of upstream platform behaviour, with no
multi-event aggregate to fold across; each event already has
final cost, tokens, and actor when it arrives.

Bolting them onto `trace-processing` would force one of:

1. Activity events get represented as fake single-span traces
   (lossy and confusing: trace\_summaries would mix gateway-proxied
   traces and per-event activity rows under the same TenantId).
2. trace\_summaries grows a discriminator column and the fold
   projection becomes branchy.

Both make trace-processing harder to reason about and add coupling
between independently-evolving subsystems. A dedicated
`activity-monitor-processing` pipeline keeps each surface's
aggregate semantics clean.

## Aggregate identity

```text theme={null}
aggregateType:  "activity_event"
aggregateId:    EventId  (one event = one aggregate, no fold across events)
tenantId:       IngestionSource.id  (matches gateway_activity_events.TenantId)
```

The fold projection (`anomalyWindow`) aggregates *across* aggregates
within a tenant, keyed by tenant and rolling window, rather than
folding events *into* one aggregate. Trace-processing folds spans
into a trace summary; this fold tallies per-tenant rolling spend,
request count, and per-actor breakdown for the past N minutes or
hours. Same machinery, different aggregate semantics.

## Slicing the redesign

The redesign ships as four slices, C0 through C3:

### C0: this doc and spec updates

* This architecture doc.
* `specs/ai-gateway/governance/anomaly-detection.feature` updated to
  drop poller language; subscriber framing throughout.
* `AnomalyAlert` Prisma model + migration `20260427020000_add_anomaly_alert/`
  doc-comment updated to reference the subscriber as producer.
* Existing receivers continue to write CH directly until C1 lands.
  This slice is doc-only so the team can review the architecture
  before more code moves.

### C1: receiver → event\_log → projection subscriber

* New event schema: `ActivityEventReceived` with the OCSF-normalised
  ActivityEventRow shape.
* New command: `RecordActivityEventCommand` wired into the
  pipeline.
* Refactor `/api/ingest/otel/:sourceId` and
  `/api/ingest/webhook/:sourceId` to call the command instead of
  writing CH directly.
* Map projection `activityEventStorage` writes to
  `gateway_activity_events` (replaces today's direct insert).
* Manual check: curl → 202 → row visible in CH (same as today, just via
  event-sourced path).

### C2: AnomalyAlert and anomaly subscriber for one rule type

* Apply the AnomalyAlert migration that's already drafted but
  doesn't ship behaviour yet.
* Add `anomalyWindow` fold projection (per-tenant rolling totals).
* Add `anomalyDetection` subscriber for `spend_spike` only first
  (cleanest mapping to the existing CostUSD field).
* Wire into `api.activityMonitor.recentAnomalies` (replaces current
  `[]` stub).
* Manual check: create a rule in the anomaly rules UI → curl a violating
  event → alert appears on `/governance` within \~30s.

### C3: Dispatch destinations

* Generic webhook + log-only first (matches PR #3351's
  triggerActionDispatch shape).
* Slack / PagerDuty / SIEM / email follow as per-destination
  adapter slices once the subscriber pattern is proven.

## What we keep from the v0 receiver code

* `IngestionSourceService` (CRUD and auth): unchanged.
* `gateway_activity_events` CH schema (migration `00019_*`):
  unchanged. The map projection writes the same columns.
* OTel and webhook normalisers (`normalizers/otel.ts` etc.):
  unchanged. The map projection calls them now instead of the
  receiver handler.
* All receiver auth, sourceId-mismatch, and 24h secret rotation
  grace: unchanged.

## What we drop from the v0 receiver code

* The direct `ActivityEventRepository.insert(...)` call from the
  receiver handler. The receiver instead enqueues an event into
  the pipeline; the map projection does the actual CH insert.
* The poller-based AnomalyEvaluatorService design that was sketched
  but never shipped. Replaced by the anomaly subscriber.

## Test strategy per slice

| Slice     | BDD spec                                                          | Integration test                                                   | Manual check                                        |
| --------- | ----------------------------------------------------------------- | ------------------------------------------------------------------ | --------------------------------------------------- |
| C0 (this) | anomaly-detection.feature updated                                 | n/a (doc + schema)                                                 | architecture review in-channel                      |
| C1        | activity-monitor pipeline scenarios in `activity-monitor.feature` | pipeline test: append event → projection fires → CH row            | curl → 202 → CH SELECT                              |
| C2        | spend\_spike scenario in anomaly-detection.feature                | subscriber test: violating fold state → AnomalyAlert.upsert called | UI rule + violating event → /governance shows alert |
| C3        | dispatch scenarios in anomaly-detection.feature                   | subscriber test: dispatch helper called with right shape           | webhook receives canonical body                     |

Each slice ships its own BDD and integration coverage before code
lands. The production architecture is subscriber-only: `evaluateNow`
appends a synthetic event and lets the subscriber handle it (a test
harness, not a parallel code path).

## Cross-references

* [PR #3351: event-driven trace triggers via subscriber](https://github.com/langwatch/langwatch/pull/3351)
  (the pattern this redesign learns from).
* [`anomaly-detection.feature`](https://github.com/langwatch/langwatch/blob/main/specs/ai-gateway/governance/anomaly-detection.feature):
  user-facing contract, updated for event-sourcing.
* [`anomaly-rules.feature`](https://github.com/langwatch/langwatch/blob/main/specs/ai-gateway/governance/anomaly-rules.feature):
  configuration entity (already shipped, unchanged).
* [`activity-monitor.feature`](https://github.com/langwatch/langwatch/blob/main/specs/ai-gateway/governance/activity-monitor.feature):
  admin UI contract (already shipped; pipeline section adds in C1).
* [Governance architecture](/docs/ai-gateway/governance/architecture): top-level governance
  architecture. This doc is the activity-monitor deep-dive linked
  from the "Activity Monitor (Tier C/D)" block.