0.8.0-dev5: first sensory publisher — SNMP link-state → NATS → :ReflexEvent

[stevenca]

Summary

Closes the loop on the dev1–dev4 foundation by wiring the first end-to-end pipeline through the agentic architecture:

SNMP poll  →  NATS subject  →  link_down reflex handler  →  Neo4j :ReflexEvent

Until now the NATS bus has been live (since dev4 deploy) but idle: no publishers, no consumers acting. After this lands, an interface flap on a real device produces a :ReflexEvent in Neo4j within seconds.

What changed

New abstractions (all swappable via Protocol)

• ReflexEventSink Protocol — persistence target for reflex outcomes (idempotent record, close).
• netcortex.episodic package — new "episodic memory" layer in the brain metaphor.
  • InMemoryReflexEventSink — test impl, dedup via stable sha256 of (handler, subject, target, occurred_at).
  • Neo4jReflexEventSink — production impl, writes :ReflexEvent nodes via shared driver with optional :AFFECTS edge to matched :Device. MERGE on stable id → at-least-once safe.
• netcortex.thalamus.SensoryPublisher — thin facade over EventBus. Validates source against SENSORY_SOURCES at construction, builds validated sensory_subject() per publish, injects source/recorded_at defaults, swallows bus failures so the poll loop never crashes.

Wiring

• ReflexContext gains event_sink field; runner persists every outcome with sink-failure isolation (a flaky sink can't kill dispatch).
• SnmpAdapter._poll_device computes ifOperStatus deltas vs prior Neo4j state and publishes one event per transition. First-observation of an interface only emits for currently-down (prevents 96-port switch event storms on first poll).
• Worker startup constructs the bus + runner + publisher via _start_reflex_pipeline; clean drain on shutdown; graceful no-op when NATS_URL is unset.

Tests

• Contract suite for ReflexEventSink (7 cases): well-formed record, idempotency on identity fields, payload-only-difference collapses, different occurred_at separates, idempotent close, record-after-close raises, concurrent record from many tasks.
• SensoryPublisher unit tests (5 cases): source validation, default injection, bus-failure swallowing, caller override of source/recorded_at.
• SNMP link-state publisher unit tests (8 cases): no-publisher / no-driver / empty-map no-ops, no-transition silence, up→down, down→up, first-observation, Neo4j-read failure, unknown oper_status.
• Runner persistence + sink-failure-isolation tests.

Test plan

• CI: all 8 jobs (lint, typecheck, unit, golden, contracts, replay, security, SBOM)
• Build image localhost:32000/netcortex:0.8.0-dev5
• helm upgrade on cpn-ful-netcortex1
• Verify worker log shows worker.reflex_pipeline_started with all 3 handlers
• After first SNMP poll cycle, look for snmp.link_state.transitions_published in worker logs (will be zero if no flaps in the window — that's expected)
• On any real interface flap, verify :ReflexEvent node appears in Neo4j:

  MATCH (e:ReflexEvent {handler: 'link_down'})
  WHERE e.recorded_at > timestamp() - 86400000
  RETURN e.subject, e.target, e.observed_at_ms, e.outcome
  ORDER BY e.observed_at_ms DESC LIMIT 20

Why this matters

• First operational use of the NATS bus.
• Establishes the pattern for every future publisher (Meraki webhook in dev6, SNMP traps in dev7, gNMI dial-out later): each is now ~50 lines, not an architectural change.
• When dev6 lands a second publisher for the same event class, the DedupStore from dev3 will get its first real workout collapsing duplicate observations.

Made with Cursor