IngestPipeline

The IngestPipeline is the engine that turns 'a thing the connector saw' into 'a thing the SocPipeline can reason about.' Without it, every connector would ship its own shape and the gate would be unable to compare a Wazuh alert to a syslog 4xx without bespoke per-connector glue. With it, every connector hands the SocPipeline an identically-shaped CanonicalEvent and the rest of the engine is normalizer-agnostic.

Dedupe matters a lot in practice. A Wazuh API poll re-emits open alerts on every poll; a Vector substrate replay-on-restart re-emits the file-tail buffer. Without a per-connector dedupe ledger, every restart would double-process every event. The IngestPipeline owns that ledger and gives the rest of the engine an idempotency guarantee on event-by-event basis.

The pipeline is also where each connector's identity is stamped onto the event. The CanonicalEvent carries `source_connector` + `event_type` + `event_id`, and those fields are what the suppression matcher, the prefilter, and the gate all key on. Get the ingest layer wrong and the whole engine misroutes; get it right and every downstream engine reads the same fields the same way.

Each connector wires its own IngestPipeline at runtime with three pieces: a Normalizer (the connector-specific parser that turns the raw event into a CanonicalEvent), an EvidenceLake handle (where the raw bytes get archived under a content-addressed key), and a DedupeIndex (per-connector SQLite table keyed on event_id).

On `pipeline.process(raw_event)` the normalizer runs first. If the normalizer raises (malformed event, unknown schema, missing required field), the pipeline returns `REJECTED` with the typed error in the audit row. The raw bytes are still written to the dead-letter side of the evidence lake so an operator can replay the event after fixing the normalizer.

If normalization succeeds, the CanonicalEvent gets a deterministic `event_id` derived from `(source_connector, raw_bytes_sha256, observed_at)`. The pipeline checks the dedupe index for that id; if it exists, the call returns `DUPLICATE` and the SocPipeline skips downstream work. If new, the pipeline writes the raw bytes to the evidence lake, registers the id in the dedupe index, and returns `ACCEPTED` with the CanonicalEvent for the SocPipeline to pick up.

The per-connector design is deliberate. A bug in the Wazuh normalizer cannot corrupt the Unifi dedupe ledger; an evidence-lake write failure on syslog does not block the M365 poll. Each connector's IngestPipeline is independently failure-isolated and independently observable.

Stable — the IngestPipeline shape has been frozen since 2026-04-30. Every shipped connector (Wazuh, Unifi, M365, syslog-substrate, plus the breadth catalog of nginx / apache / pfSense / Suricata / etc.) consumes the same interface.

Shipped 2026-06-07: `cybercli system dedupe-compact --max-age-days 90` (with `--dry-run` for safe preview) + `DedupeIndex.compact()` Python API. The Python method takes either an explicit `older_than_iso` cutoff or a `max_age_seconds` convenience; both run under the same lock as ingest so concurrent writes don't race. Remaining gap is a scheduler that fires this on a cadence — v1.x backlog.

Known limit: the per-connector failure-isolation guarantee is process-local — if the entire syslog-ingest service crashes, every connector wired into it stops. A multi-process supervisor (one process per connector kind) is on the v1.5 roadmap for high-availability installs.