Merge pull request #5 from sharedjourney/feature/event-stream

feat(event/stream): add channel-based ONVIF event consumer

.devcontainer/Dockerfile

@@ -0,0 +1,8 @@
+FROM mcr.microsoft.com/devcontainers/go:1.24-bookworm
+
+# Install node environment
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+    nodejs \
+    npm \
+    && rm -rf /var/lib/apt/lists/*

.devcontainer/devcontainer.json

@@ -0,0 +1,15 @@
+// For format details, see https://aka.ms/devcontainer.json. For config options, see the
+// README at: https://github.com/devcontainers/templates/tree/main/src/python
+{
+	"name": "go:1.24.2-bookworm",
+	"dockerFile": "Dockerfile",
+	"customizations": {
+		"vscode": {
+			"extensions": [
+				"GitHub.copilot",
+				"golang.go",
+				"GitHub.vscode-pull-request-github"
+			]
+		}
+	}
+}

.vscode/settings.json

@@ -1,24 +1,6 @@
 {
     "editor.tabSize": 2,
     "extensions.ignoreRecommendations": true,
-    "workbench.colorCustomizations": {
-      "activityBar.background": "#759570",
-      "activityBar.activeBorder": "#cfd3ef",
-      "activityBar.foreground": "#e7e7e7",
-      "activityBar.hoverBackground": "#352cea",
-      "activityBar.inactiveForeground": "#e7e7e799",
-      "activityBarBadge.background": "#cfd3ef",
-      "activityBarBadge.foreground": "#15202b",
-      "titleBar.activeBackground": "#5e7959",
-      "titleBar.inactiveBackground": "#5e795999",
-      "titleBar.activeForeground": "#e7e7e7",
-      "titleBar.inactiveForeground": "#e7e7e799",
-      "statusBarItem.hoverBackground": "#352cea",
-      "statusBar.foreground": "#e7e7e7",
-      "panel.border": "#759570",
-      "sideBar.border": "#759570",
-      "editorGroup.border": "#759570"
-    },
     "go.languageServerFlags": [],
     "go.lintOnSave": "file",
     "go.vetOnSave": "package",

docs/Development.md

@@ -32,3 +32,23 @@ python3 python/gen_commands.py
 
 > **Note:** You can also typically run the generator within your IDE thanks to the `//go:generate` lines 
 > towards the top of the `types.go` files.
+
+## Higher-level helpers
+
+Some web service directories ship hand-written, higher-level helpers
+built on top of the wire-layer commands. These are normal Go packages
+— **not** covered by the `gen_commands.py` workflow above and not
+expected to be regenerated.
+
+- [event/stream](../event/stream) — channel-based event consumer that
+  owns the pull-point subscription lifecycle (Create, Pull, Renew,
+  Unsubscribe, reconnect with jittered backoff) and decodes
+  notifications into normalized typed Events. Vendor topic strings
+  (AXIS, Hikvision, Avigilon, Hanwha, Bosch, Dahua) are classified
+  into a small set of `Kind` values. See the package `doc.go` for the
+  public surface and usage.
+- [event/topic](../event/topic) — topic identifier helpers.
+
+When adding a similar higher-level helper, place it under the relevant
+web service directory as a sub-package so consumers find it next to
+the wire-layer types it builds on.

event/stream/decode.go

@@ -0,0 +1,96 @@
+package stream
+
+import (
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/onvif/event"
+)
+
+// decode converts a single ONVIF NotificationMessage into a normalized
+// Event. Topic, Source and Data are always populated even when Kind is
+// KindUnknown so consumers can fall back to the wire form.
+func decode(msg event.NotificationMessage, deviceID string, observedAt time.Time) Event {
+	topic := string(msg.Topic.TopicKinds)
+	desc := msg.Message.Message
+	return Event{
+		Kind:       Classify(topic),
+		State:      extractState(desc.Data.SimpleItem),
+		Operation:  parsePropertyOperation(string(desc.PropertyOperation)),
+		DeviceID:   deviceID,
+		Source:     simpleItemsToMap(desc.Source.SimpleItem),
+		Data:       simpleItemsToMap(desc.Data.SimpleItem),
+		Topic:      topic,
+		Timestamp:  observedAt,
+		DeviceTime: parseDeviceTime(string(desc.UtcTime)),
+	}
+}
+
+// simpleItemsToMap returns nil for an empty list so empty notifications
+// do not allocate.
+func simpleItemsToMap(items []event.SimpleItem) map[string]string {
+	if len(items) == 0 {
+		return nil
+	}
+	m := make(map[string]string, len(items))
+	for _, it := range items {
+		m[string(it.Name)] = string(it.Value)
+	}
+	return m
+}
+
+// extractState scans Data items for a boolean-like value, returning the
+// first match. Returns StateUnknown for edge-triggered topics like
+// LineDetector/Crossed whose Data carries only an ObjectId.
+func extractState(items []event.SimpleItem) State {
+	for _, it := range items {
+		switch strings.ToLower(strings.TrimSpace(string(it.Value))) {
+		case "true", "1", "active":
+			return StateActive
+		case "false", "0", "inactive":
+			return StateInactive
+		}
+	}
+	return StateUnknown
+}
+
+// parsePropertyOperation returns PropertyUnknown for absent (optional
+// per WS-Notification) or unrecognised values.
+func parsePropertyOperation(s string) PropertyOperation {
+	switch s {
+	case "Initialized":
+		return PropertyInitialized
+	case "Changed":
+		return PropertyChanged
+	case "Deleted":
+		return PropertyDeleted
+	default:
+		return PropertyUnknown
+	}
+}
+
+// parseDeviceTime parses wsnt:UtcTime, returning the zero time when
+// absent or unparseable. Real cameras emit several flavours: with /
+// without sub-seconds, colon or compact ("+0200") offsets, and some
+// older Hikvision firmwares omit the timezone entirely (treated as
+// UTC per WS-BaseNotification which mandates UTC for UtcTime).
+func parseDeviceTime(s string) time.Time {
+	if s == "" {
+		return time.Time{}
+	}
+	for _, layout := range deviceTimeLayouts {
+		if t, err := time.Parse(layout, s); err == nil {
+			return t.UTC()
+		}
+	}
+	return time.Time{}
+}
+
+var deviceTimeLayouts = []string{
+	time.RFC3339Nano,
+	time.RFC3339,
+	"2006-01-02T15:04:05.999-0700", // Geovision
+	"2006-01-02T15:04:05-0700",     // some Dahua
+	"2006-01-02T15:04:05.999",
+	"2006-01-02T15:04:05", // older Hikvision (no timezone)
+}

event/stream/decode_test.go

@@ -0,0 +1,350 @@
+package stream
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kerberos-io/onvif/event"
+	"github.com/kerberos-io/onvif/xsd"
+	"github.com/stretchr/testify/assert"
+)
+
+// msg builds a NotificationMessage from the topic and a (PropertyOperation,
+// UtcTime, source items, data items) tuple so tests stay short and intent
+// is visible at the call site.
+func msg(topic, propOp, utcTime string, source, data map[string]string) event.NotificationMessage {
+	toItems := func(m map[string]string) []event.SimpleItem {
+		if len(m) == 0 {
+			return nil
+		}
+		items := make([]event.SimpleItem, 0, len(m))
+		for k, v := range m {
+			items = append(items, event.SimpleItem{
+				Name:  xsd.AnyType(k),
+				Value: xsd.AnyType(v),
+			})
+		}
+		return items
+	}
+	return event.NotificationMessage{
+		Topic: event.Topic{TopicKinds: xsd.String(topic)},
+		Message: event.MessageBody{
+			Message: event.MessageDescription{
+				PropertyOperation: xsd.AnyType(propOp),
+				UtcTime:           xsd.AnyType(utcTime),
+				Source:            event.Source{SimpleItem: toItems(source)},
+				Data:              event.Data{SimpleItem: toItems(data)},
+			},
+		},
+	}
+}
+
+func TestDecode_MotionActive(t *testing.T) {
+	observedAt := time.Date(2026, 5, 21, 10, 30, 1, 0, time.UTC)
+	in := msg(
+		"tns1:RuleEngine/CellMotionDetector/Motion",
+		"Changed",
+		"2026-05-21T10:30:00Z",
+		map[string]string{
+			"VideoSourceConfigurationToken": "VideoSourceConfigToken0",
+			"Rule":                          "MyMotionRule",
+		},
+		map[string]string{"IsMotion": "true"},
+	)
+
+	ev := decode(in, "axis-cam-01", observedAt)
+
+	assert.Equal(t, KindMotion, ev.Kind)
+	assert.Equal(t, StateActive, ev.State)
+	assert.Equal(t, PropertyChanged, ev.Operation)
+	assert.Equal(t, "axis-cam-01", ev.DeviceID)
+	assert.Equal(t, "VideoSourceConfigToken0", ev.Source["VideoSourceConfigurationToken"])
+	assert.Equal(t, "MyMotionRule", ev.Source["Rule"])
+	assert.Equal(t, "true", ev.Data["IsMotion"])
+	assert.Equal(t, "tns1:RuleEngine/CellMotionDetector/Motion", ev.Topic)
+	assert.True(t, ev.Timestamp.Equal(observedAt))
+	assert.Equal(t, time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC), ev.DeviceTime)
+}
+
+func TestDecode_MotionInactive(t *testing.T) {
+	in := msg(
+		"tns1:VideoSource/MotionAlarm",
+		"Changed",
+		"",
+		nil,
+		map[string]string{"State": "false"},
+	)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, KindMotion, ev.Kind)
+	assert.Equal(t, StateInactive, ev.State)
+}
+
+func TestDecode_HanwhaNumericMotionValue(t *testing.T) {
+	// Hanwha emits xsd:string values "0"/"1" instead of xsd:boolean.
+	in := msg(
+		"tns1:VideoAnalytics/tnssamsung:MotionDetection",
+		"Changed",
+		"",
+		nil,
+		map[string]string{"Motion": "1"},
+	)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, KindMotion, ev.Kind)
+	assert.Equal(t, StateActive, ev.State)
+}
+
+func TestDecode_AvigilonActiveLiteral(t *testing.T) {
+	// Avigilon and a handful of older firmwares emit "active"/"inactive"
+	// as the Data value rather than a boolean.
+	in := msg(
+		"tns1:Device/tns1:Trigger/tns1:Relay",
+		"Changed",
+		"",
+		map[string]string{"RelayToken": "Relay-1"},
+		map[string]string{"LogicalState": "active"},
+	)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, KindDigitalOutput, ev.Kind)
+	assert.Equal(t, StateActive, ev.State)
+	assert.Equal(t, "Relay-1", ev.Source["RelayToken"])
+}
+
+func TestDecode_AxisObjectAnalyticsMultiItem(t *testing.T) {
+	// AOA emits active + classType + confidence in the same Data list.
+	// The decoder must preserve every item; State picks the first
+	// boolean-like value, which is 'active'.
+	in := msg(
+		"tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1",
+		"Changed",
+		"",
+		map[string]string{"Source": "device1Scene1"},
+		map[string]string{
+			"active":     "1",
+			"classType":  "Human",
+			"confidence": "92",
+		},
+	)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, KindObjectDetected, ev.Kind)
+	assert.Equal(t, StateActive, ev.State)
+	assert.Equal(t, "Human", ev.Data["classType"])
+	assert.Equal(t, "92", ev.Data["confidence"])
+	assert.Equal(t, "1", ev.Data["active"])
+}
+
+func TestDecode_LineDetectorCrossedHasNoState(t *testing.T) {
+	// Edge-triggered topic — Data carries ObjectId, not a boolean. State
+	// must remain Unknown so consumers do not misread it as level-Active.
+	in := msg(
+		"tns1:RuleEngine/LineDetector/Crossed",
+		"Changed",
+		"",
+		map[string]string{"VideoSourceConfigurationToken": "vsct0", "Rule": "LineRule"},
+		map[string]string{"ObjectId": "42"},
+	)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, KindObjectDetected, ev.Kind)
+	assert.Equal(t, StateUnknown, ev.State)
+	assert.Equal(t, "42", ev.Data["ObjectId"])
+}
+
+func TestDecode_UnknownTopicStillPreservesWireData(t *testing.T) {
+	// Kind unknown does not mean discard: consumers may want to log or
+	// route on the raw topic when classification misses.
+	in := msg(
+		"tns1:UserAlarm/IVA",
+		"",
+		"",
+		nil,
+		map[string]string{"Custom": "true"},
+	)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, KindUnknown, ev.Kind)
+	assert.Equal(t, "tns1:UserAlarm/IVA", ev.Topic)
+	assert.Equal(t, "true", ev.Data["Custom"])
+}
+
+func TestDecode_PropertyOperationVariants(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want PropertyOperation
+	}{
+		{"initialized", "Initialized", PropertyInitialized},
+		{"changed", "Changed", PropertyChanged},
+		{"deleted", "Deleted", PropertyDeleted},
+		{"absent", "", PropertyUnknown},
+		{"unrecognised", "Bogus", PropertyUnknown},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			in := msg("tns1:VideoSource/MotionAlarm", tc.in, "", nil, nil)
+			ev := decode(in, "dev", time.Now())
+			assert.Equal(t, tc.want, ev.Operation)
+		})
+	}
+}
+
+func TestDecode_DeviceTimeParsing(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want time.Time
+	}{
+		{"rfc3339_utc", "2026-05-21T10:30:00Z", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
+		{"rfc3339_with_offset", "2026-05-21T12:30:00+02:00", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
+		{"rfc3339_subsecond", "2026-05-21T10:30:00.500Z", time.Date(2026, 5, 21, 10, 30, 0, 500_000_000, time.UTC)},
+		{"absent", "", time.Time{}},
+		{"unparseable", "not-a-date", time.Time{}},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			in := msg("tns1:VideoSource/MotionAlarm", "Changed", tc.in, nil, nil)
+			ev := decode(in, "dev", time.Now())
+			if tc.want.IsZero() {
+				assert.True(t, ev.DeviceTime.IsZero(), "DeviceTime=%v", ev.DeviceTime)
+			} else {
+				assert.True(t, ev.DeviceTime.Equal(tc.want), "got=%v want=%v", ev.DeviceTime, tc.want)
+			}
+		})
+	}
+}
+
+func TestDecode_EmptySourceAndDataYieldNilMaps(t *testing.T) {
+	// Matches the zero-value contract in types_test.go: callers can
+	// safely len() and index into Source/Data without nil-checking, but
+	// we do not allocate an empty map for empty notifications.
+	in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", nil, nil)
+	ev := decode(in, "dev", time.Now())
+	assert.Nil(t, ev.Source)
+	assert.Nil(t, ev.Data)
+}
+
+func TestDecode_StateValueIsCaseInsensitive(t *testing.T) {
+	tests := []struct {
+		name  string
+		value string
+		want  State
+	}{
+		{"true_lower", "true", StateActive},
+		{"true_upper", "TRUE", StateActive},
+		{"true_mixed", "True", StateActive},
+		{"false_lower", "false", StateInactive},
+		{"false_mixed", "False", StateInactive},
+		{"active_mixed", "Active", StateActive},
+		{"inactive_mixed", "Inactive", StateInactive},
+		{"one", "1", StateActive},
+		{"zero", "0", StateInactive},
+		{"empty", "", StateUnknown},
+		{"nonsense", "maybe", StateUnknown},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
+				nil, map[string]string{"State": tc.value})
+			ev := decode(in, "dev", time.Now())
+			assert.Equal(t, tc.want, ev.State)
+		})
+	}
+}
+
+// --- Edge cases for state extraction and time parsing ----------------
+
+func TestDecode_PropertyOperationIsCaseSensitive(t *testing.T) {
+	// Per WS-Notification §3.3 PropertyOperation values are
+	// 'Initialized' / 'Changed' / 'Deleted'. Lowercased forms are
+	// malformed and should fall through to PropertyUnknown.
+	in := msg("tns1:VideoSource/MotionAlarm", "changed", "", nil, nil)
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, PropertyUnknown, ev.Operation)
+}
+
+func TestDecode_StateValueTrimsWhitespace(t *testing.T) {
+	tests := []struct {
+		name  string
+		value string
+		want  State
+	}{
+		{"leading_trailing", "  true  ", StateActive},
+		{"tab_newline", "\ttrue\n", StateActive},
+		{"only_spaces", "   ", StateUnknown},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
+				nil, map[string]string{"State": tc.value})
+			ev := decode(in, "dev", time.Now())
+			assert.Equal(t, tc.want, ev.State)
+		})
+	}
+}
+
+func TestDecode_SimpleItemEmptyValueIsUnknownState(t *testing.T) {
+	in := msg("tns1:VideoSource/MotionAlarm", "Changed", "",
+		nil, map[string]string{"State": ""})
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, StateUnknown, ev.State)
+	v, ok := ev.Data["State"]
+	assert.True(t, ok)
+	assert.Equal(t, "", v)
+}
+
+func TestDecode_DeviceTimeAdditionalLayouts(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want time.Time
+	}{
+		{"compact_offset", "2026-05-21T12:30:00+0200", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
+		{"compact_offset_subsec", "2026-05-21T12:30:00.500+0200", time.Date(2026, 5, 21, 10, 30, 0, 500_000_000, time.UTC)},
+		{"naked_no_tz", "2026-05-21T10:30:00", time.Date(2026, 5, 21, 10, 30, 0, 0, time.UTC)},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			in := msg("tns1:VideoSource/MotionAlarm", "Changed", tc.in, nil, nil)
+			ev := decode(in, "dev", time.Now())
+			assert.True(t, ev.DeviceTime.Equal(tc.want),
+				"input=%q got=%v want=%v", tc.in, ev.DeviceTime, tc.want)
+		})
+	}
+}
+
+func TestDecode_DeviceTimeStillRejectsNonsense(t *testing.T) {
+	for _, s := range []string{"hello", "2026-13-45T99:99:99", strings.Repeat("9", 50)} {
+		in := msg("tns1:VideoSource/MotionAlarm", "Changed", s, nil, nil)
+		ev := decode(in, "dev", time.Now())
+		assert.True(t, ev.DeviceTime.IsZero(), "input=%q should yield zero, got %v", s, ev.DeviceTime)
+	}
+}
+
+// --- First-boolean-wins with explicit slice order --------------------
+
+type pair struct{ k, v string }
+
+func simpleItemsFromPairs(pairs []pair) []event.SimpleItem {
+	out := make([]event.SimpleItem, len(pairs))
+	for i, p := range pairs {
+		out[i] = event.SimpleItem{
+			Name:  xsd.AnyType(p.k),
+			Value: xsd.AnyType(p.v),
+		}
+	}
+	return out
+}
+
+func TestExtractState_FirstBooleanLikeWins(t *testing.T) {
+	// Documented behaviour: when multiple Data items have boolean-like
+	// values, the first by slice order wins. Use explicit slice
+	// construction so the assertion does not depend on map iteration
+	// order.
+	in := msg("tns1:VideoSource/MotionAlarm", "Changed", "", nil, nil)
+	in.Message.Message.Data.SimpleItem = simpleItemsFromPairs([]pair{
+		{"ObjectId", "42"},
+		{"State", "true"},
+		{"Trailer", "false"},
+	})
+	ev := decode(in, "dev", time.Now())
+	assert.Equal(t, StateActive, ev.State,
+		"first boolean-like value (State=true) must win, not Trailer=false")
+}

event/stream/doc.go

@@ -0,0 +1,28 @@
+// Package stream is a typed, channel-based consumer for ONVIF device
+// events. It hides the SOAP/XML, pull-point subscription lifecycle,
+// subscription renewal and vendor-specific topic conventions behind a
+// single Event stream.
+//
+// # Usage
+//
+//	dev, _ := onvif.NewDevice(onvif.DeviceParams{Xaddr: "...", Username: "...", Password: "..."})
+//	s, err := stream.NewStream(ctx, dev, stream.Options{DeviceID: "front-door"})
+//	if err != nil { /* construction failed: auth, network, or no event support */ }
+//	defer s.Close()
+//
+//	for ev := range s.Events() {
+//	    switch ev.Kind {
+//	    case stream.KindMotion:
+//	        if ev.State == stream.StateActive { /* start recording */ }
+//	    }
+//	}
+//
+// NewStream performs network I/O so auth and reachability failures
+// surface synchronously. Events and Errors close when the Stream stops;
+// Errors sends are non-blocking so a stalled consumer drops older
+// errors rather than blocking the pull loop. After a silent reconnect,
+// the next batch's events carry Event.AfterReconnect=true.
+//
+// See topics.go for the verified topic→Kind mapping across AXIS,
+// Hikvision, Avigilon, Hanwha, Bosch and Dahua.
+package stream

event/stream/helpers_test.go

@@ -0,0 +1,21 @@
+package stream
+
+import (
+	"testing"
+	"time"
+)
+
+// waitFor polls cond at 10ms intervals up to d. Fails the test with msg
+// if cond never returns true. Centralises the pattern that appears in
+// renew/reconnect/stream tests so retries are uniform.
+func waitFor(t *testing.T, d time.Duration, msg string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(d)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("waitFor timed out after %s: %s", d, msg)
+}

event/stream/reconnect.go

@@ -0,0 +1,108 @@
+package stream
+
+import (
+	"context"
+	"math/rand"
+	"time"
+)
+
+// maxRecreateBackoff caps exponential backoff between recreate
+// attempts. Sized for fleet deployments: at 30s a 1000-camera setup
+// recovering from a switch reboot would generate sustained
+// reconnect traffic; 5 minutes lets the network settle.
+const maxRecreateBackoff = 5 * time.Minute
+
+// jitterFraction prevents thundering-herd reconnects when many
+// cameras drop together (switch reboot, NAT timeout).
+const jitterFraction = 0.25
+
+// pullLoop runs PullMessages → decode → Events. After
+// ReconnectAfterFailures consecutive errors it asks attemptRecreate
+// to rebuild the subscription. The next batch's events carry
+// AfterReconnect=true so consumers can suppress the Initialized
+// replay ONVIF emits on a new subscription.
+func (s *Stream) pullLoop(ctx context.Context) {
+	var failures int
+	recreateBackoff := s.opts.RetryBackoff
+	var afterReconnect bool
+
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+		msgs, err := pullMessages(s.caller, s.getPullPoint(), s.opts)
+		if err != nil {
+			s.surfaceError(ErrPullFailed{Err: err})
+			failures++
+			if !s.opts.DisableReconnect && failures >= s.opts.ReconnectAfterFailures {
+				justRecreated, cont := s.attemptRecreate(ctx, &failures, &recreateBackoff)
+				if !cont {
+					return
+				}
+				if justRecreated {
+					afterReconnect = true
+				}
+				continue
+			}
+			if !sleepCtx(ctx, s.opts.RetryBackoff) {
+				return
+			}
+			continue
+		}
+		failures = 0
+		recreateBackoff = s.opts.RetryBackoff
+		observedAt := s.now()
+		for _, m := range msgs {
+			ev := decode(m, s.opts.DeviceID, observedAt)
+			if afterReconnect {
+				ev.AfterReconnect = true
+				// Clear once the camera transitions past the
+				// Initialized replay to live events.
+				if ev.Operation != PropertyInitialized {
+					afterReconnect = false
+				}
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case s.events <- ev:
+			}
+		}
+	}
+}
+
+// attemptRecreate returns (justRecreated, cont). cont is false only
+// when ctx cancelled during backoff so the caller exits the loop.
+func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) (justRecreated, cont bool) {
+	addr, err := createPullPoint(s.caller, s.opts)
+	if err != nil {
+		s.surfaceError(ErrRecreateFailed{Err: err})
+		if !sleepCtx(ctx, jitter(*backoff)) {
+			return false, false
+		}
+		*backoff *= 2
+		if *backoff > maxRecreateBackoff {
+			*backoff = maxRecreateBackoff
+		}
+		return false, true
+	}
+	s.setPullPoint(addr)
+	*failures = 0
+	*backoff = s.opts.RetryBackoff
+	return true, true
+}
+
+// jitter perturbs d by ±jitterFraction so synchronised drops do not
+// produce a synchronised reconnect surge.
+func jitter(d time.Duration) time.Duration {
+	if d <= 0 {
+		return time.Nanosecond
+	}
+	spread := float64(d) * jitterFraction
+	delta := (rand.Float64()*2 - 1) * spread
+	out := time.Duration(float64(d) + delta)
+	if out <= 0 {
+		out = time.Nanosecond
+	}
+	return out
+}

event/stream/reconnect_test.go

@@ -0,0 +1,315 @@
+package stream
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// createPullPointRespAlt mirrors the first fixture but returns a
+// different SubscriptionReference Address so a test can prove that
+// subsequent pulls hit the recreated endpoint.
+const createPullPointRespAlt = `<?xml version="1.0" encoding="UTF-8"?>
+<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+              xmlns:wsa="http://www.w3.org/2005/08/addressing"
+              xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body>
+    <tev:CreatePullPointSubscriptionResponse>
+      <tev:SubscriptionReference>
+        <wsa:Address>http://camera.local/onvif/Events/PullSub_2</wsa:Address>
+      </tev:SubscriptionReference>
+      <tev:CurrentTime>2026-05-21T10:30:10Z</tev:CurrentTime>
+      <tev:TerminationTime>2026-05-21T10:31:10Z</tev:TerminationTime>
+    </tev:CreatePullPointSubscriptionResponse>
+  </env:Body>
+</env:Envelope>`
+
+// --- Recreate after pull failures ------------------------------------
+
+func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueCallMethod(createPullPointRespAlt, nil)
+
+	fc.queueSendSoap("", errors.New("transient failure"))
+	fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		DeviceID:               "cam-1",
+		PullTimeout:            50 * time.Millisecond,
+		ReconnectAfterFailures: 1,
+		RetryBackoff:           10 * time.Millisecond,
+		InitialTermination:     30 * time.Second,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	ev := receive(t, s.Events(), 2*time.Second)
+	assert.Equal(t, KindMotion, ev.Kind)
+
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	require.Len(t, fc.callMethodCalls, 2,
+		"expected exactly 2 CallMethod calls (initial + recreate)")
+	var newEndpointPulls int
+	for _, c := range fc.sendSoapCalls {
+		if c[0] == "http://camera.local/onvif/Events/PullSub_2" {
+			newEndpointPulls++
+		}
+	}
+	assert.GreaterOrEqual(t, newEndpointPulls, 1,
+		"expected pulls against the recreated subscription endpoint")
+}
+
+func TestStream_BackoffWhenRecreateFails(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.mu.Lock()
+	fc.defaultCall = fakeResp{err: errors.New("recreate fail")}
+	fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:            10 * time.Millisecond,
+		ReconnectAfterFailures: 1,
+		RetryBackoff:           10 * time.Millisecond,
+		InitialTermination:     30 * time.Second,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(2 * time.Second)
+	var calls atomic.Int32
+	for time.Now().Before(deadline) {
+		fc.mu.Lock()
+		calls.Store(int32(len(fc.callMethodCalls)))
+		fc.mu.Unlock()
+		if calls.Load() >= 4 {
+			break
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	assert.GreaterOrEqual(t, calls.Load(), int32(4),
+		"expected stream to retry recreate (>=3 retries on top of the initial create)")
+}
+
+func TestStream_ReconnectAfterFailuresDefault(t *testing.T) {
+	o := defaultOptions()
+	assert.Equal(t, 3, o.ReconnectAfterFailures)
+}
+
+func TestStream_RetryBackoffDefault(t *testing.T) {
+	o := defaultOptions()
+	assert.Equal(t, time.Second, o.RetryBackoff)
+}
+
+func TestStream_DisableReconnectKeepsRetryingOriginalEndpoint(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.mu.Lock()
+	fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:        10 * time.Millisecond,
+		RetryBackoff:       10 * time.Millisecond,
+		InitialTermination: 30 * time.Second,
+		DisableReconnect:   true,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	time.Sleep(200 * time.Millisecond)
+	fc.mu.Lock()
+	calls := len(fc.callMethodCalls)
+	fc.mu.Unlock()
+	assert.Equal(t, 1, calls, "DisableReconnect must prevent recreate; got %d CallMethod calls", calls)
+}
+
+func TestStream_RecreateResetsFailuresAndBackoffOnSuccess(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueCallMethod(createPullPointRespAlt, nil)
+	fc.queueSendSoap("", errors.New("first failure"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:            10 * time.Millisecond,
+		ReconnectAfterFailures: 1,
+		RetryBackoff:           10 * time.Millisecond,
+		InitialTermination:     30 * time.Second,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	time.Sleep(200 * time.Millisecond)
+	fc.mu.Lock()
+	calls := len(fc.callMethodCalls)
+	fc.mu.Unlock()
+	assert.Equal(t, 2, calls,
+		"after one failure + successful recreate, no further recreates expected; got %d", calls)
+}
+
+func TestStream_PullPointMutationVisibleToRenewLoopUnderRace(t *testing.T) {
+	// Drives the pullPoint write-by-pullLoop / read-by-renewLoop race
+	// so -race actually exercises the mutex critical sections.
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	for i := 0; i < 50; i++ {
+		fc.queueCallMethod(createPullPointRespAlt, nil)
+	}
+	fc.mu.Lock()
+	fc.defaultSendSoap = fakeResp{err: errors.New("recurring pull fail")}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:            5 * time.Millisecond,
+		ReconnectAfterFailures: 1,
+		RetryBackoff:           1 * time.Millisecond,
+		InitialTermination:     20 * time.Millisecond,
+		RenewMargin:            2 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	time.Sleep(300 * time.Millisecond)
+}
+
+// --- Typed errors from the reconnect path ----------------------------
+
+func TestStream_PullErrorIsTypedErrPullFailed(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueSendSoap("", errors.New("transient"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:        50 * time.Millisecond,
+		RetryBackoff:       10 * time.Millisecond,
+		InitialTermination: 30 * time.Second,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	select {
+	case e := <-s.Errors():
+		var pullErr ErrPullFailed
+		require.True(t, errors.As(e, &pullErr), "expected ErrPullFailed, got %T: %v", e, e)
+		assert.Contains(t, pullErr.Err.Error(), "transient")
+	case <-time.After(time.Second):
+		t.Fatal("expected ErrPullFailed on Errors channel")
+	}
+}
+
+func TestStream_RecreateErrorIsTypedErrRecreateFailed(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.mu.Lock()
+	fc.defaultCall = fakeResp{err: errors.New("recreate fail")}
+	fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:            10 * time.Millisecond,
+		ReconnectAfterFailures: 1,
+		RetryBackoff:           10 * time.Millisecond,
+		InitialTermination:     30 * time.Second,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(time.Second)
+	var sawRecreate bool
+	for time.Now().Before(deadline) && !sawRecreate {
+		select {
+		case e := <-s.Errors():
+			var rec ErrRecreateFailed
+			if errors.As(e, &rec) {
+				sawRecreate = true
+				assert.Contains(t, rec.Err.Error(), "recreate fail")
+			}
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+	assert.True(t, sawRecreate, "expected at least one ErrRecreateFailed on Errors")
+}
+
+func TestStream_AfterReconnectFlagSetOnFirstPostRecreateBatch(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueCallMethod(createPullPointRespAlt, nil)
+
+	fc.queueSendSoap("", errors.New("transient"))
+	fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
+	fc.queueSendSoap(pullMessagesResp(motionMsg("false")), nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:            50 * time.Millisecond,
+		ReconnectAfterFailures: 1,
+		RetryBackoff:           10 * time.Millisecond,
+		InitialTermination:     30 * time.Second,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	ev1 := receive(t, s.Events(), 2*time.Second)
+	assert.True(t, ev1.AfterReconnect, "first event after recreate must carry AfterReconnect=true")
+	assert.Equal(t, StateActive, ev1.State)
+
+	ev2 := receive(t, s.Events(), 2*time.Second)
+	assert.False(t, ev2.AfterReconnect, "subsequent events should not carry AfterReconnect")
+	assert.Equal(t, StateInactive, ev2.State)
+}
+
+// --- Jitter ----------------------------------------------------------
+
+func TestJitter_StaysWithinFraction(t *testing.T) {
+	const base = time.Second
+	low := time.Duration(float64(base) * (1 - jitterFraction))
+	high := time.Duration(float64(base) * (1 + jitterFraction))
+	for i := 0; i < 200; i++ {
+		got := jitter(base)
+		assert.GreaterOrEqual(t, got, low, "iteration %d", i)
+		assert.LessOrEqual(t, got, high, "iteration %d", i)
+	}
+}
+
+func TestJitter_ZeroAndNegativeReturnPositive(t *testing.T) {
+	assert.Greater(t, jitter(0), time.Duration(0))
+	assert.Greater(t, jitter(-time.Second), time.Duration(0))
+}
+
+func TestJitter_VariesAcrossCalls(t *testing.T) {
+	first := jitter(time.Second)
+	allEqual := true
+	for i := 0; i < 10; i++ {
+		if jitter(time.Second) != first {
+			allEqual = false
+			break
+		}
+	}
+	assert.False(t, allEqual, "jitter is producing a constant; rand seed not working")
+}
+
+func TestMaxRecreateBackoff_Is5Minutes(t *testing.T) {
+	assert.Equal(t, 5*time.Minute, maxRecreateBackoff)
+}

event/stream/renew.go

@@ -0,0 +1,58 @@
+package stream
+
+import (
+	"context"
+	"encoding/xml"
+	"fmt"
+	"time"
+
+	"github.com/kerberos-io/onvif/event"
+	"github.com/kerberos-io/onvif/xsd"
+)
+
+// renewLoop surfaces renew failures and continues. A permanently
+// failing renew lets the subscription die at the camera; the pull
+// loop's reconnect path then recreates it — recreate is the only
+// reliable recovery once a subscription is GC'd.
+func (s *Stream) renewLoop(ctx context.Context) {
+	interval := s.opts.InitialTermination - s.opts.RenewMargin
+	if interval <= 0 {
+		// Pathological config (margin >= termination): renew at
+		// half termination so we still refresh.
+		interval = s.opts.InitialTermination / 2
+		if interval <= 0 {
+			interval = time.Second
+		}
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := renewPullPoint(s.caller, s.getPullPoint(), s.opts); err != nil {
+				s.surfaceError(ErrRenewFailed{Err: err})
+			}
+		}
+	}
+}
+
+// renewPullPoint sends Renew with an absolute UTC TerminationTime.
+// WS-BaseNotification §6.1.1 also allows xsd:duration but older
+// Hikvision, some Dahua and some Bosch firmwares reject the
+// relative form.
+func renewPullPoint(c caller, endpoint string, opts Options) error {
+	absoluteEnd := time.Now().UTC().Add(opts.InitialTermination).Format("2006-01-02T15:04:05Z")
+	req := event.Renew{TerminationTime: xsd.String(absoluteEnd)}
+	body, err := xml.Marshal(req)
+	if err != nil {
+		return fmt.Errorf("marshal Renew: %w", err)
+	}
+	resp, err := c.SendSoap(endpoint, string(body))
+	if err != nil {
+		return err
+	}
+	_, err = readClose(resp)
+	return err
+}

event/stream/renew_test.go

@@ -0,0 +1,170 @@
+package stream
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// countSendSoapMatching counts how many recorded SendSoap calls have a
+// body containing needle. Safe to call concurrently with the run loop.
+func countSendSoapMatching(fc *fakeCaller, needle string) int {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	n := 0
+	for _, c := range fc.sendSoapCalls {
+		if strings.Contains(c[1], needle) {
+			n++
+		}
+	}
+	return n
+}
+
+func TestStream_RenewsSubscriptionBeforeExpiry(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// 100 ms termination with 10 ms margin -> renew every ~90 ms.
+	s, err := newStream(ctx, fc, Options{
+		DeviceID:           "cam-1",
+		InitialTermination: 100 * time.Millisecond,
+		RenewMargin:        10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(500 * time.Millisecond)
+	var renewCount int
+	for time.Now().Before(deadline) {
+		renewCount = countSendSoapMatching(fc, "Renew")
+		if renewCount >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	assert.GreaterOrEqual(t, renewCount, 1, "expected at least one Renew SendSoap call within 500ms")
+}
+
+func TestStream_RenewSendsToSubscriptionEndpoint(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s, err := newStream(ctx, fc, Options{
+		InitialTermination: 80 * time.Millisecond,
+		RenewMargin:        10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		if countSendSoapMatching(fc, "Renew") >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	var renewEndpoint string
+	for _, c := range fc.sendSoapCalls {
+		if strings.Contains(c[1], "Renew") {
+			renewEndpoint = c[0]
+			break
+		}
+	}
+	require.NotEmpty(t, renewEndpoint, "no Renew call found")
+	assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", renewEndpoint,
+		"Renew must target the SubscriptionReference Address")
+}
+
+func TestStream_RenewMarginAppliesDefault(t *testing.T) {
+	o := defaultOptions()
+	assert.Equal(t, 10*time.Second, o.RenewMargin)
+}
+
+func TestStream_RenewErrorSurfacedOnErrorsChannel(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	// Defaults return empty pulls indefinitely so the pull loop is clean.
+	// Override defaultSendSoap on the fly to return a Renew error for
+	// any body that looks like a Renew. We do that by tagging the
+	// default response with an err, then resetting after capturing one.
+	// Simpler: just queue several explicit Renew-error responses; the
+	// fake's queue is consumed in FIFO and the pull body never matches
+	// 'Renew', so queued errors will land on the renew call only if
+	// queued before any pulls. To bias the order we drain via a custom
+	// default.
+	fc.mu.Lock()
+	fc.defaultSendSoap = fakeResp{err: errInjected{}}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s, err := newStream(ctx, fc, Options{
+		InitialTermination: 80 * time.Millisecond,
+		RenewMargin:        10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	select {
+	case e := <-s.Errors():
+		assert.Contains(t, e.Error(), "injected")
+	case <-time.After(time.Second):
+		t.Fatal("expected an error on Errors channel from failing Renew/pull")
+	}
+}
+
+// errInjected is a sentinel error type so the test message has a stable
+// substring without depending on a wrapped string match.
+type errInjected struct{}
+
+func (errInjected) Error() string { return "injected fake error" }
+
+func TestRenew_SendsAbsoluteDateTimeNotDuration(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		InitialTermination: 30 * time.Millisecond,
+		RenewMargin:        5 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		if countSendSoapMatching(fc, "Renew") >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	var renewBody string
+	for _, c := range fc.sendSoapCalls {
+		if strings.Contains(c[1], "Renew") {
+			renewBody = c[1]
+			break
+		}
+	}
+	require.NotEmpty(t, renewBody, "no Renew call observed")
+	// Absolute form is "YYYY-MM-DDTHH:MM:SSZ" not "PTnS".
+	assert.NotContains(t, renewBody, "PT", "Renew should not send relative duration; some firmwares reject it")
+	assert.Regexp(t, `\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z`, renewBody, "Renew should send absolute RFC3339 UTC")
+}

event/stream/soap.go

@@ -0,0 +1,178 @@
+package stream
+
+import (
+	"bytes"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/kerberos-io/onvif/event"
+	"github.com/kerberos-io/onvif/xsd"
+)
+
+// maxResponseBytes caps SOAP response buffering. ONVIF PullMessages
+// bodies are normally <100KB even with dense analytics payloads;
+// 10 MiB is comfortably above legitimate traffic while keeping a
+// hostile or buggy camera from OOMing the process.
+const maxResponseBytes = 10 << 20
+
+func createPullPoint(c caller, opts Options) (string, error) {
+	term := xsd.String(durationToXSD(opts.InitialTermination))
+	req := event.CreatePullPointSubscription{InitialTerminationTime: &term}
+	if opts.RawTopicFilter != "" {
+		req.Filter = &event.FilterType{
+			TopicExpression: &event.TopicExpressionType{
+				Dialect:    xsd.String("http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet"),
+				TopicKinds: xsd.String(opts.RawTopicFilter),
+			},
+		}
+	}
+	resp, err := c.CallMethod(req)
+	if err != nil {
+		return "", err
+	}
+	body, err := readClose(resp)
+	if err != nil {
+		return "", err
+	}
+	var decoded event.CreatePullPointSubscriptionResponse
+	if err := unmarshalNode(body, "CreatePullPointSubscriptionResponse", &decoded); err != nil {
+		return "", err
+	}
+	addr := string(decoded.SubscriptionReference.Address)
+	if addr == "" {
+		return "", errors.New("CreatePullPointSubscription response has empty SubscriptionReference Address")
+	}
+	return addr, nil
+}
+
+// pullMessages returns an empty slice (no error) when the camera had
+// nothing within PullTimeout.
+func pullMessages(c caller, endpoint string, opts Options) ([]event.NotificationMessage, error) {
+	req := event.PullMessages{
+		Timeout:      xsd.Duration(durationToXSD(opts.PullTimeout)),
+		MessageLimit: xsd.Int(opts.MessageLimit),
+	}
+	body, err := xml.Marshal(req)
+	if err != nil {
+		return nil, fmt.Errorf("marshal PullMessages: %w", err)
+	}
+	resp, err := c.SendSoap(endpoint, string(body))
+	if err != nil {
+		return nil, err
+	}
+	respBody, err := readClose(resp)
+	if err != nil {
+		return nil, err
+	}
+	var decoded event.PullMessagesResponse
+	if err := unmarshalNode(respBody, "PullMessagesResponse", &decoded); err != nil {
+		return nil, err
+	}
+	return decoded.NotificationMessage, nil
+}
+
+// unsubscribePullPoint is best-effort. Empty endpoint is a no-op
+// (construction failed before installing one).
+func unsubscribePullPoint(c caller, endpoint string) error {
+	if endpoint == "" {
+		return nil
+	}
+	body, err := xml.Marshal(event.Unsubscribe{})
+	if err != nil {
+		return fmt.Errorf("marshal Unsubscribe: %w", err)
+	}
+	resp, err := c.SendSoap(endpoint, string(body))
+	if err != nil {
+		return err
+	}
+	_, err = readClose(resp)
+	return err
+}
+
+func readClose(resp *http.Response) (string, error) {
+	if resp == nil || resp.Body == nil {
+		return "", errors.New("nil HTTP response")
+	}
+	defer resp.Body.Close()
+	b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
+	if err != nil {
+		return "", fmt.Errorf("read response body: %w", err)
+	}
+	return string(b), nil
+}
+
+// unmarshalNode finds the first XML start element with the given local
+// name and decodes it into out. ONVIF SOAP responses are wrapped in an
+// envelope with many namespace prefixes; keying on local name only
+// sidesteps namespace matching.
+//
+// When the camera returns a SOAP Fault, the fault reason is returned
+// as the error so callers can distinguish auth / expired-subscription
+// from "unparseable response".
+func unmarshalNode(body, localName string, out any) error {
+	if reason := extractSOAPFault(body); reason != "" {
+		return fmt.Errorf("ONVIF SOAP fault: %s", reason)
+	}
+	dec := xml.NewDecoder(bytes.NewBufferString(body))
+	for {
+		tok, err := dec.Token()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				return fmt.Errorf("ONVIF response missing %s element", localName)
+			}
+			return fmt.Errorf("scan ONVIF response: %w", err)
+		}
+		start, ok := tok.(xml.StartElement)
+		if !ok {
+			continue
+		}
+		if start.Name.Local != localName {
+			continue
+		}
+		if err := dec.DecodeElement(out, &start); err != nil {
+			return fmt.Errorf("decode %s: %w", localName, err)
+		}
+		return nil
+	}
+}
+
+var (
+	// SOAP 1.1: <faultstring>reason</faultstring>
+	soap11FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?faultstring[^>]*>(.*?)</(?:[^:>\s]+:)?faultstring>`)
+	// SOAP 1.2: <Fault>...<Reason><Text>reason</Text></Reason>...</Fault>
+	soap12FaultRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Reason\b[^>]*>.*?<(?:[^:>\s]+:)?Text[^>]*>(.*?)</(?:[^:>\s]+:)?Text>`)
+)
+
+// extractSOAPFault returns the reason text from a SOAP fault or empty
+// when the body is not a fault. Handles SOAP 1.1 (faultstring) and
+// SOAP 1.2 (Reason/Text) shapes.
+func extractSOAPFault(body string) string {
+	if !strings.Contains(body, "Fault") {
+		return ""
+	}
+	if m := soap11FaultRE.FindStringSubmatch(body); len(m) > 1 {
+		return strings.TrimSpace(m[1])
+	}
+	if m := soap12FaultRE.FindStringSubmatch(body); len(m) > 1 {
+		return strings.TrimSpace(m[1])
+	}
+	return ""
+}
+
+// durationToXSD formats a duration as xsd:duration PTnS. Second
+// precision is sufficient — ONVIF cameras do not honour sub-second
+// pull timeouts.
+func durationToXSD(d time.Duration) string {
+	secs := int(d.Round(time.Second).Seconds())
+	if secs <= 0 {
+		secs = 1
+	}
+	return "PT" + strconv.Itoa(secs) + "S"
+}

event/stream/soap_test.go

@@ -0,0 +1,86 @@
+package stream
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// --- SOAP fault detection ---------------------------------------------
+
+func TestExtractSOAPFault_SOAP11(t *testing.T) {
+	body := `<?xml version="1.0"?>
+<env:Envelope xmlns:env="http://schemas.xmlsoap.org/soap/envelope/">
+  <env:Body>
+    <env:Fault>
+      <faultcode>env:Client</faultcode>
+      <faultstring>The action requested requires authorization and the sender is not authorized</faultstring>
+    </env:Fault>
+  </env:Body>
+</env:Envelope>`
+	got := extractSOAPFault(body)
+	assert.Contains(t, got, "not authorized")
+}
+
+func TestExtractSOAPFault_SOAP12(t *testing.T) {
+	body := `<?xml version="1.0"?>
+<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope">
+  <env:Body>
+    <env:Fault>
+      <env:Code><env:Value>env:Sender</env:Value></env:Code>
+      <env:Reason><env:Text xml:lang="en">Subscription has expired</env:Text></env:Reason>
+    </env:Fault>
+  </env:Body>
+</env:Envelope>`
+	got := extractSOAPFault(body)
+	assert.Contains(t, got, "Subscription has expired")
+}
+
+func TestExtractSOAPFault_NotAFault(t *testing.T) {
+	assert.Empty(t, extractSOAPFault(createPullPointResp))
+}
+
+func TestExtractSOAPFault_EmptyBody(t *testing.T) {
+	assert.Empty(t, extractSOAPFault(""))
+}
+
+func TestUnmarshalNode_ReturnsFaultReasonInsteadOfMissingElement(t *testing.T) {
+	body := `<env:Envelope xmlns:env="http://schemas.xmlsoap.org/soap/envelope/">
+  <env:Body><env:Fault><faultstring>not authorized</faultstring></env:Fault></env:Body>
+</env:Envelope>`
+	var out struct{}
+	err := unmarshalNode(body, "PullMessagesResponse", &out)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "not authorized")
+	assert.NotContains(t, err.Error(), "missing PullMessagesResponse")
+}
+
+// --- Bounded body read -----------------------------------------------
+
+func TestReadClose_LimitsBodySize(t *testing.T) {
+	if maxResponseBytes < 1024 {
+		t.Skip("limit too small for this test")
+	}
+	big := strings.Repeat("A", maxResponseBytes+1024)
+	body := "<env:Envelope><env:Body>" + big + "</env:Body></env:Envelope>"
+	fc := newFakeCaller()
+	fc.queueCallMethod(body, nil)
+	// Construction will fail because the truncated body has no
+	// CreatePullPointSubscriptionResponse — that's fine; what matters is
+	// the read completes without OOM.
+	_, err := newStream(testContext(t), fc, Options{})
+	assert.Error(t, err)
+}
+
+// testContext returns a Background context already wired to cancel via
+// t.Cleanup so the test does not need to manage the cancellation
+// goroutine inline.
+func testContext(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	return ctx
+}

event/stream/stream.go

@@ -0,0 +1,284 @@
+package stream
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// closeDrainTimeout bounds Close's wait for the pull and renew
+// goroutines to exit. The loops block in caller.SendSoap which is not
+// ctx-aware (the underlying http.Client is the only thing that can
+// unblock them — see caller below). On a hung HTTP transport Close
+// would otherwise wait forever; instead it returns an error and lets
+// the calling agent move on.
+const closeDrainTimeout = 5 * time.Second
+
+// closeUnsubscribeTimeout bounds the Unsubscribe SOAP call issued by
+// Close. A subscription expires at the camera once InitialTermination
+// elapses without a renew, so a missed unsubscribe is at worst
+// cosmetic.
+const closeUnsubscribeTimeout = 5 * time.Second
+
+// Options configures a Stream.
+//
+// Zero-value policy: every duration / int field treats zero as "use
+// the default". To opt out of reconnect set DisableReconnect=true
+// (ReconnectAfterFailures=0 would otherwise collide with the default
+// injection). For unbuffered Events / Errors channels set
+// BufferSize=-1.
+type Options struct {
+	DeviceID string
+	// RawTopicFilter is the ONVIF ConcreteSet TopicExpression filter
+	// passed verbatim to CreatePullPointSubscription. Callers should
+	// normally leave this empty and rely on Classify for routing —
+	// server-side filtering is fragile across vendors and empty is
+	// required for AXIS.
+	RawTopicFilter string
+	// PullTimeout — zero means default (5s).
+	PullTimeout time.Duration
+	// MessageLimit — zero means default (32). Busy AXIS cameras with
+	// many configured rules can burst beyond 10 per pull.
+	MessageLimit int
+	// InitialTermination — zero means default (60s).
+	InitialTermination time.Duration
+	// RenewMargin — larger margins tolerate slower networks at the
+	// cost of more renew calls. Zero means default (10s).
+	RenewMargin time.Duration
+	// ReconnectAfterFailures — pull-points die for many reasons
+	// (camera reboot, subscription GC after a renew miss, NAT
+	// timeout); rebuilding the subscription is the only reliable
+	// recovery. Zero means default (3). Set DisableReconnect=true
+	// to disable.
+	ReconnectAfterFailures int
+	// DisableReconnect makes the pull loop retry against the
+	// original endpoint until ctx is cancelled.
+	DisableReconnect bool
+	// RetryBackoff is the base sleep between pull/recreate failures.
+	// Recreate failures double this up to maxRecreateBackoff. Zero
+	// means default (1s).
+	RetryBackoff time.Duration
+	// BufferSize — zero means default (16); use -1 for unbuffered.
+	BufferSize int
+}
+
+func defaultOptions() Options {
+	return Options{
+		PullTimeout:            5 * time.Second,
+		MessageLimit:           32,
+		InitialTermination:     60 * time.Second,
+		RenewMargin:            10 * time.Second,
+		ReconnectAfterFailures: 3,
+		RetryBackoff:           time.Second,
+		BufferSize:             16,
+	}
+}
+
+func (o Options) withDefaults() Options {
+	d := defaultOptions()
+	if o.PullTimeout > 0 {
+		d.PullTimeout = o.PullTimeout
+	}
+	if o.MessageLimit > 0 {
+		d.MessageLimit = o.MessageLimit
+	}
+	if o.InitialTermination > 0 {
+		d.InitialTermination = o.InitialTermination
+	}
+	if o.RenewMargin > 0 {
+		d.RenewMargin = o.RenewMargin
+	}
+	if o.ReconnectAfterFailures > 0 {
+		d.ReconnectAfterFailures = o.ReconnectAfterFailures
+	}
+	if o.RetryBackoff > 0 {
+		d.RetryBackoff = o.RetryBackoff
+	}
+	switch {
+	case o.BufferSize > 0:
+		d.BufferSize = o.BufferSize
+	case o.BufferSize < 0:
+		d.BufferSize = 0
+	}
+	d.DeviceID = o.DeviceID
+	d.RawTopicFilter = o.RawTopicFilter
+	d.DisableReconnect = o.DisableReconnect
+	return d
+}
+
+// caller is the *onvif.Device subset Stream depends on. Implementations
+// must:
+//
+//   - Be safe for concurrent use — pull and renew goroutines call in
+//     from separate goroutines. *onvif.Device satisfies this via
+//     http.Client.
+//   - Enforce a per-request timeout via the underlying HTTP client.
+//     The methods do not take a ctx, so ctx-cancel cannot interrupt a
+//     hung request; only the HTTP client's own timeout can. Close
+//     bounds its drain wait at closeDrainTimeout to survive a misbehaving
+//     caller, but a leaking goroutine remains until the HTTP call
+//     eventually returns.
+type caller interface {
+	CallMethod(method any) (*http.Response, error)
+	SendSoap(endpoint, body string) (*http.Response, error)
+}
+
+type deviceCaller struct{ dev *onvif.Device }
+
+func (d deviceCaller) CallMethod(m any) (*http.Response, error) {
+	return d.dev.CallMethod(m)
+}
+
+func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) {
+	return d.dev.SendSoap(endpoint, body)
+}
+
+// Stream owns a single ONVIF pull-point subscription. Safe for Close
+// from any goroutine while readers consume Events / Errors. Close is
+// idempotent.
+type Stream struct {
+	caller caller
+	opts   Options
+
+	pullPointMu sync.Mutex
+	pullPoint   string
+
+	events chan Event
+	errors chan error
+
+	cancel context.CancelFunc
+	done   chan struct{}
+
+	closeOnce sync.Once
+	closeErr  error
+
+	// now is overridable so tests can make timestamps deterministic.
+	now func() time.Time
+}
+
+func (s *Stream) getPullPoint() string {
+	s.pullPointMu.Lock()
+	defer s.pullPointMu.Unlock()
+	return s.pullPoint
+}
+
+func (s *Stream) setPullPoint(addr string) {
+	s.pullPointMu.Lock()
+	defer s.pullPointMu.Unlock()
+	s.pullPoint = addr
+}
+
+// NewStream creates a Stream and performs CreatePullPointSubscription
+// synchronously so connectivity and authentication failures surface
+// from NewStream rather than landing on Errors later.
+//
+// The returned Stream stops when ctx is cancelled or Close is called.
+func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) {
+	return newStream(ctx, deviceCaller{dev: dev}, opts)
+}
+
+func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) {
+	opts = opts.withDefaults()
+	addr, err := createPullPoint(c, opts)
+	if err != nil {
+		return nil, fmt.Errorf("create pull point subscription: %w", err)
+	}
+	runCtx, cancel := context.WithCancel(ctx)
+	s := &Stream{
+		caller:    c,
+		opts:      opts,
+		pullPoint: addr,
+		events:    make(chan Event, opts.BufferSize),
+		errors:    make(chan error, opts.BufferSize),
+		cancel:    cancel,
+		done:      make(chan struct{}),
+		now:       time.Now,
+	}
+	go s.run(runCtx)
+	return s, nil
+}
+
+// Events returns the channel of decoded notifications. Closed when
+// the Stream stops.
+func (s *Stream) Events() <-chan Event { return s.events }
+
+// Errors returns the channel of non-fatal errors. Sends are
+// non-blocking; consumers that fall behind drop older errors. Closed
+// when the Stream stops.
+func (s *Stream) Errors() <-chan error { return s.errors }
+
+// Close stops the background goroutines, waits up to closeDrainTimeout
+// for them to exit, and then Unsubscribes from the camera (also bounded,
+// by closeUnsubscribeTimeout). Subsequent calls are no-ops.
+//
+// If the drain times out the goroutines are likely wedged inside a
+// non-ctx-aware caller.SendSoap; they will exit on their own once the
+// HTTP call returns. Unsubscribe is skipped in that case — the
+// subscription expires at the camera anyway.
+func (s *Stream) Close() error {
+	s.closeOnce.Do(func() {
+		s.cancel()
+
+		select {
+		case <-s.done:
+		case <-time.After(closeDrainTimeout):
+			s.closeErr = fmt.Errorf("close: pull/renew loops did not drain within %s (likely stuck in caller HTTP)", closeDrainTimeout)
+			return
+		}
+
+		errCh := make(chan error, 1)
+		go func() {
+			errCh <- unsubscribePullPoint(s.caller, s.getPullPoint())
+		}()
+		select {
+		case err := <-errCh:
+			if err != nil {
+				s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err)
+			}
+		case <-time.After(closeUnsubscribeTimeout):
+			s.closeErr = fmt.Errorf("unsubscribe pull point: timeout after %s", closeUnsubscribeTimeout)
+		}
+	})
+	return s.closeErr
+}
+
+func (s *Stream) run(ctx context.Context) {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		s.renewLoop(ctx)
+	}()
+	s.pullLoop(ctx)
+	wg.Wait()
+
+	// Explicit close order after both goroutines have exited so a
+	// future maintainer extending this function does not rely on
+	// defer-ordering for channel-close safety.
+	close(s.errors)
+	close(s.events)
+	close(s.done)
+}
+
+func (s *Stream) surfaceError(err error) {
+	select {
+	case s.errors <- err:
+	default:
+	}
+}
+
+// sleepCtx returns false if ctx was cancelled, true if d elapsed.
+func sleepCtx(ctx context.Context, d time.Duration) bool {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-t.C:
+		return true
+	}
+}

event/stream/stream_test.go

@@ -0,0 +1,497 @@
+package stream
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// --- fakeCaller --------------------------------------------------------
+
+// fakeCaller is a test double for the caller interface. Each method
+// returns the next queued response; when the queue is exhausted it falls
+// back to a default response so the indefinite pull loop does not
+// require tests to enumerate every call.
+//
+// blockUnsubscribe, when non-nil, causes SendSoap calls whose body
+// contains "Unsubscribe" to block until the channel is closed.
+// blockAllSendSoap, when non-nil, blocks every SendSoap call until
+// closed (simulates a hung HTTP transport).
+type fakeCaller struct {
+	mu               sync.Mutex
+	callMethodResps  []fakeResp
+	sendSoapResps    []fakeResp
+	defaultSendSoap  fakeResp
+	defaultCall      fakeResp
+	callMethodCalls  []any
+	sendSoapCalls    [][2]string
+	blockUnsubscribe chan struct{}
+	blockAllSendSoap chan struct{}
+}
+
+type fakeResp struct {
+	body string
+	err  error
+}
+
+func newFakeCaller() *fakeCaller {
+	return &fakeCaller{
+		// Default: indefinite empty pulls, indefinite OK unsubscribes.
+		defaultSendSoap: fakeResp{body: pullMessagesResp()},
+		defaultCall:     fakeResp{err: errors.New("fakeCaller: no default CallMethod response")},
+	}
+}
+
+func (f *fakeCaller) queueCallMethod(body string, err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.callMethodResps = append(f.callMethodResps, fakeResp{body: body, err: err})
+}
+
+func (f *fakeCaller) queueSendSoap(body string, err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.sendSoapResps = append(f.sendSoapResps, fakeResp{body: body, err: err})
+}
+
+func (f *fakeCaller) CallMethod(m any) (*http.Response, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.callMethodCalls = append(f.callMethodCalls, m)
+	r := f.defaultCall
+	if len(f.callMethodResps) > 0 {
+		r = f.callMethodResps[0]
+		f.callMethodResps = f.callMethodResps[1:]
+	}
+	if r.err != nil {
+		return nil, r.err
+	}
+	return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil
+}
+
+func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) {
+	f.mu.Lock()
+	f.sendSoapCalls = append(f.sendSoapCalls, [2]string{endpoint, body})
+	r := f.defaultSendSoap
+	if len(f.sendSoapResps) > 0 {
+		r = f.sendSoapResps[0]
+		f.sendSoapResps = f.sendSoapResps[1:]
+	}
+	block := f.blockUnsubscribe
+	blockAll := f.blockAllSendSoap
+	f.mu.Unlock()
+
+	if blockAll != nil {
+		<-blockAll
+	}
+	if block != nil && strings.Contains(body, "Unsubscribe") {
+		<-block
+	}
+
+	if r.err != nil {
+		return nil, r.err
+	}
+	return &http.Response{Body: io.NopCloser(strings.NewReader(r.body))}, nil
+}
+
+func (f *fakeCaller) sendSoapCallCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.sendSoapCalls)
+}
+
+// --- fixture SOAP envelopes -------------------------------------------
+
+// createPullPointResp is the minimal SOAP envelope the lib's existing
+// xml.Decoder + getXMLNode path can extract a pull-point address from.
+const createPullPointResp = `<?xml version="1.0" encoding="UTF-8"?>
+<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+              xmlns:wsa="http://www.w3.org/2005/08/addressing"
+              xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body>
+    <tev:CreatePullPointSubscriptionResponse>
+      <tev:SubscriptionReference>
+        <wsa:Address>http://camera.local/onvif/Events/PullSub_1</wsa:Address>
+      </tev:SubscriptionReference>
+      <tev:CurrentTime>2026-05-21T10:30:00Z</tev:CurrentTime>
+      <tev:TerminationTime>2026-05-21T10:31:00Z</tev:TerminationTime>
+    </tev:CreatePullPointSubscriptionResponse>
+  </env:Body>
+</env:Envelope>`
+
+func pullMessagesResp(messages ...string) string {
+	return `<?xml version="1.0" encoding="UTF-8"?>
+<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+              xmlns:tev="http://www.onvif.org/ver10/events/wsdl"
+              xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2"
+              xmlns:tt="http://www.onvif.org/ver10/schema">
+  <env:Body>
+    <tev:PullMessagesResponse>
+      <tev:CurrentTime>2026-05-21T10:30:05Z</tev:CurrentTime>
+      <tev:TerminationTime>2026-05-21T10:31:05Z</tev:TerminationTime>
+      ` + strings.Join(messages, "\n") + `
+    </tev:PullMessagesResponse>
+  </env:Body>
+</env:Envelope>`
+}
+
+func motionMsg(value string) string {
+	return `<wsnt:NotificationMessage>
+  <wsnt:Topic Dialect="http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet">tns1:RuleEngine/CellMotionDetector/Motion</wsnt:Topic>
+  <wsnt:Message>
+    <tt:Message PropertyOperation="Changed" UtcTime="2026-05-21T10:30:00Z">
+      <tt:Source>
+        <tt:SimpleItem Name="VideoSourceConfigurationToken" Value="VSC0"/>
+      </tt:Source>
+      <tt:Data>
+        <tt:SimpleItem Name="IsMotion" Value="` + value + `"/>
+      </tt:Data>
+    </tt:Message>
+  </wsnt:Message>
+</wsnt:NotificationMessage>`
+}
+
+const unsubscribeResp = `<?xml version="1.0" encoding="UTF-8"?>
+<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+              xmlns:wsnt="http://docs.oasis-open.org/wsn/b-2">
+  <env:Body>
+    <wsnt:UnsubscribeResponse/>
+  </env:Body>
+</env:Envelope>`
+
+// --- helpers -----------------------------------------------------------
+
+// receive waits up to d for an event on ch, failing the test if none
+// arrives.
+func receive(t *testing.T, ch <-chan Event, d time.Duration) Event {
+	t.Helper()
+	select {
+	case ev, ok := <-ch:
+		if !ok {
+			t.Fatalf("event channel closed before receiving")
+		}
+		return ev
+	case <-time.After(d):
+		t.Fatalf("timed out waiting for event after %s", d)
+	}
+	return Event{} // unreachable
+}
+
+// --- tests -------------------------------------------------------------
+
+func TestNewStream_CreatesPullPointAtConstruction(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	// Queue an empty pull so the run loop can spin without exploding.
+	fc.queueSendSoap(pullMessagesResp(), nil)
+	fc.queueSendSoap(unsubscribeResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	require.NoError(t, err)
+	require.NotNil(t, s)
+	require.NoError(t, s.Close())
+
+	// CreatePullPointSubscription was called exactly once.
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	require.Len(t, fc.callMethodCalls, 1, "expected one CallMethod call (CreatePullPointSubscription)")
+}
+
+func TestStream_DeliversDecodedEvents(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
+	// Provide subsequent empty pulls so the loop doesn't starve before Close.
+	fc.queueSendSoap(pullMessagesResp(), nil)
+	fc.queueSendSoap(pullMessagesResp(), nil)
+	fc.queueSendSoap(unsubscribeResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	require.NoError(t, err)
+	defer s.Close()
+
+	ev := receive(t, s.Events(), 2*time.Second)
+	assert.Equal(t, KindMotion, ev.Kind)
+	assert.Equal(t, StateActive, ev.State)
+	assert.Equal(t, "cam-1", ev.DeviceID)
+	assert.Equal(t, "tns1:RuleEngine/CellMotionDetector/Motion", ev.Topic)
+	assert.Equal(t, "VSC0", ev.Source["VideoSourceConfigurationToken"])
+	assert.Equal(t, "true", ev.Data["IsMotion"])
+}
+
+func TestStream_PullsAgainstSubscriptionAddress(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueSendSoap(pullMessagesResp(), nil)
+	fc.queueSendSoap(pullMessagesResp(), nil)
+	fc.queueSendSoap(unsubscribeResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	require.NoError(t, err)
+
+	// Wait until at least one pull happened, then close.
+	for i := 0; i < 50 && fc.sendSoapCallCount() == 0; i++ {
+		time.Sleep(10 * time.Millisecond)
+	}
+	require.NoError(t, s.Close())
+
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	require.NotEmpty(t, fc.sendSoapCalls, "expected at least one PullMessages SendSoap call")
+	endpoint := fc.sendSoapCalls[0][0]
+	assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", endpoint,
+		"PullMessages must target the SubscriptionReference Address returned by CreatePullPoint")
+	// Last call (Close) should target the same endpoint with an Unsubscribe body.
+	last := fc.sendSoapCalls[len(fc.sendSoapCalls)-1]
+	assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", last[0])
+	assert.Contains(t, last[1], "Unsubscribe")
+}
+
+func TestNewStream_ReturnsErrorWhenCreatePullPointFails(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod("", errors.New("network down"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	assert.Error(t, err)
+	assert.Nil(t, s)
+}
+
+func TestStream_ClosedContextStopsRunLoop(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	// Many empty pulls so the loop is hot when we cancel.
+	for i := 0; i < 20; i++ {
+		fc.queueSendSoap(pullMessagesResp(), nil)
+	}
+	fc.queueSendSoap(unsubscribeResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	require.NoError(t, err)
+
+	// Wait for at least one pull.
+	for i := 0; i < 50 && fc.sendSoapCallCount() == 0; i++ {
+		time.Sleep(10 * time.Millisecond)
+	}
+	cancel()
+
+	// Close should still complete cleanly; the goroutine must drain.
+	require.NoError(t, s.Close())
+
+	// Events channel must close so consumers can range-loop safely.
+	select {
+	case _, ok := <-s.Events():
+		assert.False(t, ok, "Events channel should be closed after Close()")
+	case <-time.After(time.Second):
+		t.Fatal("Events channel was not closed within 1s")
+	}
+}
+
+func TestStream_PullErrorSurfacedOnErrorsChannel(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.queueSendSoap("", errors.New("transient pull failure"))
+	// Then a clean pull so the loop keeps running.
+	fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil)
+	fc.queueSendSoap(pullMessagesResp(), nil)
+	fc.queueSendSoap(unsubscribeResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	require.NoError(t, err)
+	defer s.Close()
+
+	select {
+	case e := <-s.Errors():
+		assert.Contains(t, e.Error(), "transient pull failure")
+	case <-time.After(2 * time.Second):
+		t.Fatal("expected an error on the Errors channel")
+	}
+	// After the transient failure the loop continued and decoded.
+	ev := receive(t, s.Events(), 2*time.Second)
+	assert.Equal(t, KindMotion, ev.Kind)
+}
+
+func TestStream_OptionsApplyDefaults(t *testing.T) {
+	o := defaultOptions()
+	assert.Equal(t, 5*time.Second, o.PullTimeout)
+	assert.Equal(t, 32, o.MessageLimit)
+	assert.Equal(t, 60*time.Second, o.InitialTermination)
+	assert.Equal(t, 16, o.BufferSize)
+}
+
+func TestStream_DoesNotPanicOnPullExitingDuringClose(t *testing.T) {
+	// Regression guard: Close should not race with the run goroutine
+	// in a way that double-closes the events/errors channels.
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	for i := 0; i < 5; i++ {
+		fc.queueSendSoap(pullMessagesResp(), nil)
+	}
+	fc.queueSendSoap(unsubscribeResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{DeviceID: "cam-1"})
+	require.NoError(t, err)
+	assert.NotPanics(t, func() {
+		require.NoError(t, s.Close())
+		// Double-close should be a no-op, not a panic.
+		_ = s.Close()
+	})
+}
+
+// --- Close error / timeout paths -------------------------------------
+
+func TestClose_ReturnsUnsubscribeError(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	fc.mu.Lock()
+	fc.defaultSendSoap = fakeResp{err: errors.New("simulated transport failure")}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
+	require.NoError(t, err)
+
+	err = s.Close()
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "unsubscribe pull point")
+	assert.Contains(t, err.Error(), "simulated transport failure")
+}
+
+func TestClose_BoundedByTimeoutOnHungUnsubscribe(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	block := make(chan struct{})
+	defer close(block) // release the hung Unsubscribe so the fake's goroutine exits
+	fc.mu.Lock()
+	fc.blockUnsubscribe = block
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
+	require.NoError(t, err)
+
+	start := time.Now()
+	err = s.Close()
+	elapsed := time.Since(start)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "timeout")
+	assert.Less(t, elapsed, closeUnsubscribeTimeout+time.Second,
+		"Close exceeded bound (%s); expected ~%s", elapsed, closeUnsubscribeTimeout)
+}
+
+// --- NewStream edge cases --------------------------------------------
+
+func TestNewStream_CtxAlreadyCancelled(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // cancel before NewStream
+
+	s, err := newStream(ctx, fc, Options{InitialTermination: 30 * time.Second})
+	require.NoError(t, err)
+	require.NotNil(t, s)
+
+	select {
+	case _, ok := <-s.Events():
+		assert.False(t, ok, "events channel should be closed when ctx is pre-cancelled")
+	case <-time.After(time.Second):
+		t.Fatal("events channel was not closed within 1s")
+	}
+	_ = s.Close()
+}
+
+// --- fakeCaller self-test --------------------------------------------
+
+func TestFakeCaller_QueueThenDefaultFallback(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueSendSoap("first", nil)
+	fc.queueSendSoap("second", nil)
+
+	r1, err := fc.SendSoap("ep", "body")
+	require.NoError(t, err)
+	b1 := make([]byte, 10)
+	n, _ := r1.Body.Read(b1)
+	assert.Equal(t, "first", string(b1[:n]))
+
+	r2, _ := fc.SendSoap("ep", "body")
+	b2 := make([]byte, 10)
+	n, _ = r2.Body.Read(b2)
+	assert.Equal(t, "second", string(b2[:n]))
+
+	// Queue is exhausted; default kicks in.
+	r3, err := fc.SendSoap("ep", "body")
+	require.NoError(t, err)
+	require.NotNil(t, r3)
+	b3 := make([]byte, 2048)
+	n, _ = r3.Body.Read(b3)
+	assert.Contains(t, string(b3[:n]), "PullMessagesResponse",
+		"default SendSoap should be an empty PullMessagesResponse envelope")
+}
+
+func TestClose_BoundedWhenLoopsStuckOnHungHTTP(t *testing.T) {
+	// Simulates a hung HTTP transport: every SendSoap blocks
+	// indefinitely. The pull and renew loops are wedged inside
+	// SendSoap and ctx-cancel cannot unblock them. Close must still
+	// return within its bounded budget so the agent's shutdown does
+	// not hang.
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	blockAll := make(chan struct{})
+	defer close(blockAll)
+	fc.mu.Lock()
+	fc.blockAllSendSoap = blockAll
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:        100 * time.Millisecond,
+		InitialTermination: 30 * time.Second,
+	})
+	require.NoError(t, err)
+
+	// Wait until pullLoop is actually parked inside the blocked
+	// SendSoap. Without this, Close races with the loop's first
+	// iteration and exits via the ctx pre-check instead of
+	// exercising the drain-timeout path.
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) && fc.sendSoapCallCount() == 0 {
+		time.Sleep(10 * time.Millisecond)
+	}
+	require.GreaterOrEqual(t, fc.sendSoapCallCount(), 1, "pullLoop never reached SendSoap")
+
+	start := time.Now()
+	err = s.Close()
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "drain", "expected a drain-timeout error")
+	// Total budget is closeDrainTimeout for the wait + ~0 for unsubscribe
+	// (which is skipped when drain times out). Give plenty of slack for
+	// scheduling on a loaded CI machine.
+	assert.Less(t, elapsed, closeDrainTimeout+2*time.Second,
+		"Close exceeded bound (%s); expected ~%s", elapsed, closeDrainTimeout)
+}

event/stream/topics.go

@@ -0,0 +1,143 @@
+package stream
+
+import "strings"
+
+// Classify maps an ONVIF topic string to the normalized Kind. Returns
+// KindUnknown when no rule matches.
+//
+// The classifier strips XML-namespace prefixes from each "/"-separated
+// segment so it is robust to vendor namespaces (tns1:, tnsaxis:,
+// tnssamsung:, ...). Matching is case-sensitive — ONVIF topics are
+// case-sensitive per the spec.
+//
+// Sources cross-checked when building the rule set below:
+//   - ONVIF Topic Namespace XML
+//     https://www.onvif.org/onvif/ver10/topics/topicns.xml
+//   - ONVIF Analytics Service Spec
+//     https://www.onvif.org/specs/srv/analytics/ONVIF-VideoAnalytics-Service-Spec-v220.pdf
+//   - ONVIF Device IO Service Spec
+//     https://www.onvif.org/specs/srv/io/ONVIF-DeviceIo-Service-Spec.pdf
+//   - openvideolibs/onvif-parsers (Apache-2.0) — empirical topic table
+//     extracted from Home Assistant
+//     https://github.com/openvideolibs/onvif-parsers
+func Classify(topic string) Kind {
+	if topic == "" {
+		return KindUnknown
+	}
+	canonical := canonicalizeTopic(topic)
+	for _, rule := range topicRules {
+		if strings.Contains(canonical, rule.needle) {
+			return rule.kind
+		}
+	}
+	return KindUnknown
+}
+
+// canonicalizeTopic strips the XML-namespace prefix from each
+// "/"-separated segment, collapsing Avigilon's per-segment-prefixed
+// form ("tns1:Device/tns1:Trigger/tns1:Relay") and the plain form
+// ("tns1:Device/Trigger/Relay") to the same matchable path.
+func canonicalizeTopic(topic string) string {
+	segments := strings.Split(topic, "/")
+	for i, seg := range segments {
+		if idx := strings.Index(seg, ":"); idx >= 0 {
+			segments[i] = seg[idx+1:]
+		}
+	}
+	return strings.Join(segments, "/")
+}
+
+// topicRules is evaluated in order — first match wins. Keep more
+// specific rules ahead of broader ones. LineDetector/Crossed is
+// edge-triggered (no boolean State); the decoder leaves State as
+// StateUnknown for it.
+var topicRules = []struct {
+	needle string
+	kind   Kind
+}{
+	// tns1:VideoSource/MotionAlarm — Profile S basic motion.
+	// https://www.onvif.org/ver10/topics/topicns.xml
+	// https://developer.axis.com/vapix/network-video/event-and-action-services/
+	{"VideoSource/MotionAlarm", KindMotion},
+
+	// tns1:VideoAnalytics/MotionAlarm — Bosch publishes motion under
+	// VideoAnalytics rather than VideoSource.
+	// https://media.boschsecurity.com/fs/media/pb/media/partners_1/integration_tools_1/developer/bosch-metadata-and-iva-events.pdf
+	{"VideoAnalytics/MotionAlarm", KindMotion},
+
+	// tns1:VideoAnalytics/tnssamsung:MotionDetection — Hanwha vendor.
+	// https://github.com/home-assistant/core/issues/66493
+	{"VideoAnalytics/MotionDetection", KindMotion},
+
+	// tns1:RuleEngine/CellMotionDetector/Motion — ONVIF Analytics
+	// standard cell-motion rule.
+	// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.3
+	// https://www.hikvisioneurope.com/eu/portal/portal/Technical%20Materials/24%20How%20To/CCTV/How%20to%20solve%20third%20party%20camera%20motion%20detection%20issue.pdf
+	{"CellMotionDetector/Motion", KindMotion},
+
+	// tns1:RuleEngine/MotionRegionDetector/Motion — AXIS region rule.
+	// https://developer.axis.com/vapix/network-video/event-and-action-services/
+	{"MotionRegionDetector/Motion", KindMotion},
+
+	// AXIS Guard suite — vendor analytics apps with Camera<N>Profile<ID>
+	// suffixes. Treated as motion so they can drive motion-triggered
+	// recording on cameras using these apps instead of basic VMD.
+	// https://developer.axis.com/vapix/applications/motion-guard
+	{"CameraApplicationPlatform/MotionGuard/", KindMotion},
+	{"CameraApplicationPlatform/FenceGuard/", KindMotion},
+	{"CameraApplicationPlatform/LoiteringGuard/", KindMotion},
+
+	// tns1:RuleEngine/TamperDetector/Tamper — standard ONVIF tamper rule.
+	// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.5
+	{"TamperDetector/Tamper", KindTampering},
+
+	// tns1:VideoSource/GlobalSceneChange/ImagingService — the proper
+	// lens-cover signal on firmwares without TamperDetector.
+	// https://www.onvif.org/ver10/topics/topicns.xml
+	{"GlobalSceneChange", KindTampering},
+
+	// tns1:VideoAnalytics/tnssamsung:TamperingDetection — Hanwha.
+	// https://github.com/home-assistant/core/issues/66493
+	{"VideoAnalytics/TamperingDetection", KindTampering},
+
+	// VideoSource/ImageToo* — imaging-quality alarms. See KindImageQuality
+	// for the rationale on splitting these out from KindTampering.
+	// https://www.onvif.org/ver10/topics/topicns.xml
+	{"VideoSource/ImageTooDark", KindImageQuality},
+	{"VideoSource/ImageTooBright", KindImageQuality},
+	{"VideoSource/ImageTooBlurry", KindImageQuality},
+
+	// tns1:Device/Trigger/DigitalInput — standard. Avigilon's per-segment-
+	// prefixed serialisation ("tns1:Device/tns1:Trigger/tns1:DigitalInput")
+	// folds to the same canonical path.
+	// ONVIF-DeviceIo-Service-Spec.pdf §5.2
+	{"Trigger/DigitalInput", KindDigitalInput},
+	// ONVIF-DeviceIo-Service-Spec.pdf §5.3
+	{"Trigger/Relay", KindDigitalOutput},
+
+	// tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario<N>
+	// — Scenario suffixes are numeric per AOA configuration. Prefix-match
+	// because of the dynamic suffix.
+	// https://developer.axis.com/analytics/axis-object-analytics/how-to-guides/axis-object-analytics-counting-data/
+	{"ObjectAnalytics/", KindObjectDetected},
+
+	// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.4
+	{"LineDetector/Crossed", KindObjectDetected},
+	{"FieldDetector/ObjectsInside", KindObjectDetected},
+
+	// tns1:RuleEngine/MyRuleDetector/<RuleName> — vendor rules under the
+	// ONVIF MyRuleDetector container. Explicitly whitelisted because the
+	// same container also carries non-object rules (Bosch Counter,
+	// Occupancy) that must not classify as ObjectDetected.
+	// https://media.boschsecurity.com/fs/media/pb/media/partners_1/integration_tools_1/developer/bosch-metadata-and-iva-events.pdf
+	{"MyRuleDetector/HumanDetect", KindObjectDetected},
+	{"MyRuleDetector/VehicleDetect", KindObjectDetected},
+	{"MyRuleDetector/PeopleDetect", KindObjectDetected},
+	{"MyRuleDetector/ObjectsInside", KindObjectDetected},
+	{"MyRuleDetector/FaceDetect", KindObjectDetected},
+
+	{"Audio/DetectedSound", KindAudioAlarm},
+	// https://developer.axis.com/vapix/network-video/event-and-action-services/
+	{"AudioSource/TriggerLevel", KindAudioAlarm},
+	{"AudioAnalytics/SoundDetection", KindAudioAlarm},
+}

event/stream/topics_test.go

@@ -0,0 +1,138 @@
+package stream
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClassifyTopic(t *testing.T) {
+	tests := []struct {
+		name  string
+		topic string
+		want  Kind
+	}{
+		// --- Motion -----------------------------------------------------
+
+		{"video_source_motion_alarm", "tns1:VideoSource/MotionAlarm", KindMotion},
+		{"cell_motion_detector", "tns1:RuleEngine/CellMotionDetector/Motion", KindMotion},
+		{"motion_region_detector", "tns1:RuleEngine/MotionRegionDetector/Motion", KindMotion},
+		{"bosch_video_analytics_motion", "tns1:VideoAnalytics/MotionAlarm", KindMotion},
+		{"hanwha_samsung_motion", "tns1:VideoAnalytics/tnssamsung:MotionDetection", KindMotion},
+
+		// AXIS Guard suite — vendor analytics apps.
+		{"axis_motion_guard", "tnsaxis:CameraApplicationPlatform/MotionGuard/Camera1ProfileANY", KindMotion},
+		{"axis_fence_guard", "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1ProfileANY", KindMotion},
+		{"axis_loitering_guard", "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1ProfileANY", KindMotion},
+
+		// --- Tampering --------------------------------------------------
+
+		{"tamper_detector", "tns1:RuleEngine/TamperDetector/Tamper", KindTampering},
+		{"global_scene_change", "tns1:VideoSource/GlobalSceneChange/ImagingService", KindTampering},
+		{"hanwha_tampering", "tns1:VideoAnalytics/tnssamsung:TamperingDetection", KindTampering},
+
+		// --- Image quality (separated from Tampering) ------------------
+
+		{"image_too_dark", "tns1:VideoSource/ImageTooDark/ImagingService", KindImageQuality},
+		{"image_too_bright", "tns1:VideoSource/ImageTooBright/ImagingService", KindImageQuality},
+		{"image_too_blurry", "tns1:VideoSource/ImageTooBlurry/ImagingService", KindImageQuality},
+
+		// --- Digital input ---------------------------------------------
+
+		{"digital_input", "tns1:Device/Trigger/DigitalInput", KindDigitalInput},
+		{"digital_input_avigilon", "tns1:Device/tns1:Trigger/tns1:DigitalInput", KindDigitalInput},
+
+		// --- Digital output --------------------------------------------
+
+		{"relay", "tns1:Device/Trigger/Relay", KindDigitalOutput},
+		{"relay_avigilon", "tns1:Device/tns1:Trigger/tns1:Relay", KindDigitalOutput},
+
+		// --- Object analytics ------------------------------------------
+
+		// AXIS Object Analytics uses numeric scenario suffixes.
+		{"axis_object_analytics_scenario_1", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", KindObjectDetected},
+		{"axis_object_analytics_scenario_2", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario2", KindObjectDetected},
+
+		// Standard rule-engine analytics topics.
+		{"line_detector_crossed", "tns1:RuleEngine/LineDetector/Crossed", KindObjectDetected},
+		{"field_detector_objects_inside", "tns1:RuleEngine/FieldDetector/ObjectsInside", KindObjectDetected},
+
+		// Whitelisted MyRuleDetector sub-rules.
+		{"my_rule_detector_human", "tns1:RuleEngine/MyRuleDetector/HumanDetect", KindObjectDetected},
+		{"my_rule_detector_vehicle", "tns1:RuleEngine/MyRuleDetector/VehicleDetect", KindObjectDetected},
+		{"my_rule_detector_people", "tns1:RuleEngine/MyRuleDetector/PeopleDetect", KindObjectDetected},
+		{"my_rule_detector_face", "tns1:RuleEngine/MyRuleDetector/FaceDetect", KindObjectDetected},
+		{"my_rule_detector_objects_inside", "tns1:RuleEngine/MyRuleDetector/ObjectsInside", KindObjectDetected},
+
+		// --- Audio -----------------------------------------------------
+
+		{"audio_detected_sound", "tns1:AudioAnalytics/Audio/DetectedSound", KindAudioAlarm},
+		{"axis_audio_trigger_level", "tns1:AudioSource/tnsaxis:TriggerLevel", KindAudioAlarm},
+		{"hanwha_sound_detection", "tns1:AudioAnalytics/tnssamsung:SoundDetection", KindAudioAlarm},
+
+		// --- Negative cases --------------------------------------------
+
+		{"empty", "", KindUnknown},
+		{"unknown_topic", "tns1:UserAlarm/IVA", KindUnknown},
+		{"unrelated_recording_config", "tns1:RecordingConfig/JobState", KindUnknown},
+
+		// MyRuleDetector overmatch guard — Bosch publishes counter and
+		// occupancy under the same container and these must not be
+		// classified as object detection.
+		{"my_rule_detector_counter_not_object", "tns1:RuleEngine/MyRuleDetector/Counter", KindUnknown},
+		{"my_rule_detector_occupancy_not_object", "tns1:RuleEngine/MyRuleDetector/Occupancy", KindUnknown},
+
+		// Substring guards.
+		{"motion_recording_not_motion", "tns1:Recording/MotionRecording/Started", KindUnknown},
+		{"audio_encoder_config_not_audio_alarm", "tns1:Configuration/AudioEncoderConfiguration", KindUnknown},
+		{"relay_failure_not_digital_output", "tns1:Device/HardwareFailure/RelayFailure", KindUnknown},
+		{"digital_input_config_not_digital_input", "tns1:Device/IO/DigitalInputConfiguration", KindUnknown},
+		{"tamper_detector_log_not_tampering", "tns1:Device/Diagnostics/TamperDetectorLog", KindUnknown},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, Classify(tc.topic), "topic=%q", tc.topic)
+		})
+	}
+}
+
+func TestClassifyIsCaseSensitive(t *testing.T) {
+	// ONVIF topic identifiers are case-sensitive per the spec; a
+	// lowercased topic must not match a capitalised pattern.
+	assert.Equal(t, KindUnknown, Classify("tns1:videosource/motionalarm"))
+}
+
+func TestCanonicalizeTopicStripsNamespaces(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{"single_namespace", "tns1:VideoSource/MotionAlarm", "VideoSource/MotionAlarm"},
+		{"per_segment_namespace", "tns1:Device/tns1:Trigger/tns1:Relay", "Device/Trigger/Relay"},
+		{"vendor_namespace_inner", "tns1:VideoAnalytics/tnssamsung:MotionDetection", "VideoAnalytics/MotionDetection"},
+		{"axis_outer_namespace", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1", "CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1"},
+		{"empty", "", ""},
+		{"no_colon_passthrough", "Foo/Bar", "Foo/Bar"},
+		{"double_slash_keeps_empty_segment", "tns1://Foo", "//Foo"},
+		{"colon_only_segment_collapses_to_empty", "tns1:/Foo", "/Foo"},
+		{"trailing_colon_segment", "tns1:", ""},
+		{"multi_colon_takes_first", "tns1:Foo:Bar/Baz", "Foo:Bar/Baz"},
+		{"leading_slash_kept", "/tns1:Foo", "/Foo"},
+		{"trailing_slash_kept", "tns1:Foo/", "Foo/"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, canonicalizeTopic(tc.in), "input=%q", tc.in)
+		})
+	}
+}
+
+func TestClassifyRuleOrder_ObjectAnalyticsBeforeGenericObjects(t *testing.T) {
+	// Locks the invariant that the prefix rule "ObjectAnalytics/" is
+	// matched before the broader "ObjectsInside" rule. Without this
+	// ordering, AXIS AOA topics that contain neither would still classify
+	// correctly via the ObjectAnalytics/ rule; we encode the dependency.
+	topic := "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1Scenario1"
+	assert.Equal(t, KindObjectDetected, Classify(topic))
+}

event/stream/types.go

@@ -0,0 +1,159 @@
+package stream
+
+import (
+	"fmt"
+	"time"
+)
+
+// Kind is the normalized category of an ONVIF event, independent of the
+// camera vendor's topic naming.
+type Kind uint8
+
+const (
+	KindUnknown Kind = iota
+	KindMotion
+	KindTampering
+	// KindImageQuality covers VideoSource imaging alarms. Kept separate
+	// from KindTampering because they fire on legitimate sunset / dawn /
+	// condensation transitions, not on interference.
+	KindImageQuality
+	KindDigitalInput
+	KindDigitalOutput
+	KindObjectDetected
+	KindAudioAlarm
+)
+
+func (k Kind) String() string {
+	switch k {
+	case KindUnknown:
+		return "Unknown"
+	case KindMotion:
+		return "Motion"
+	case KindTampering:
+		return "Tampering"
+	case KindImageQuality:
+		return "ImageQuality"
+	case KindDigitalInput:
+		return "DigitalInput"
+	case KindDigitalOutput:
+		return "DigitalOutput"
+	case KindObjectDetected:
+		return "ObjectDetected"
+	case KindAudioAlarm:
+		return "AudioAlarm"
+	default:
+		return fmt.Sprintf("Kind(%d)", uint8(k))
+	}
+}
+
+// State is the active/inactive level carried by a boolean ONVIF property
+// event. StateUnknown is used both when the value cannot be parsed and
+// when the topic is edge-triggered and carries no boolean state.
+type State uint8
+
+const (
+	StateUnknown State = iota
+	StateActive
+	StateInactive
+)
+
+func (s State) String() string {
+	switch s {
+	case StateUnknown:
+		return "Unknown"
+	case StateActive:
+		return "Active"
+	case StateInactive:
+		return "Inactive"
+	default:
+		return fmt.Sprintf("State(%d)", uint8(s))
+	}
+}
+
+// PropertyOperation mirrors the wsnt:PropertyOperation attribute.
+// PropertyUnknown covers both "absent on the wire" (the attribute is
+// optional) and "unrecognised value".
+type PropertyOperation uint8
+
+const (
+	PropertyUnknown PropertyOperation = iota
+	PropertyInitialized
+	PropertyChanged
+	PropertyDeleted
+)
+
+func (p PropertyOperation) String() string {
+	switch p {
+	case PropertyUnknown:
+		return "Unknown"
+	case PropertyInitialized:
+		return "Initialized"
+	case PropertyChanged:
+		return "Changed"
+	case PropertyDeleted:
+		return "Deleted"
+	default:
+		return fmt.Sprintf("PropertyOperation(%d)", uint8(p))
+	}
+}
+
+// Event is a single normalized notification from an ONVIF device.
+//
+// Source and Data are maps because ONVIF notifications can carry
+// multiple SimpleItems — AXIS Object Analytics emits active+classType+
+// confidence in one Data list, DigitalInput carries InputToken in Source
+// and LogicalState in Data.
+type Event struct {
+	Kind      Kind
+	State     State
+	Operation PropertyOperation
+	DeviceID  string
+	Source    map[string]string
+	Data      map[string]string
+	Topic     string
+	Timestamp time.Time
+	// DeviceTime is the camera-reported wsnt:UtcTime. Cameras drift —
+	// prefer Timestamp for ordering and DeviceTime only for forensics or
+	// cross-camera correlation when the caller manages NTP.
+	DeviceTime time.Time
+	// AfterReconnect is true for events delivered after the Stream
+	// silently recreated its subscription. Cameras replay current state
+	// with PropertyInitialized on a new subscription; watch this flag to
+	// suppress duplicate edge-detection. Cleared on the first non-
+	// Initialized event.
+	AfterReconnect bool
+}
+
+// Op identifies which Stream operation failed.
+type Op string
+
+const (
+	OpPull     Op = "pull"
+	OpRenew    Op = "renew"
+	OpRecreate Op = "recreate"
+)
+
+// ErrPullFailed wraps a transient PullMessages failure. The pull loop
+// surfaces it and continues.
+type ErrPullFailed struct{ Err error }
+
+func (e ErrPullFailed) Error() string { return fmt.Sprintf("pull messages: %v", e.Err) }
+func (e ErrPullFailed) Unwrap() error { return e.Err }
+func (ErrPullFailed) Op() Op          { return OpPull }
+
+// ErrRenewFailed wraps a Renew SOAP failure. Recovered implicitly: a
+// permanently failing renew lets the subscription die, pull starts
+// failing, and the reconnect path recreates it.
+type ErrRenewFailed struct{ Err error }
+
+func (e ErrRenewFailed) Error() string { return fmt.Sprintf("renew pull point: %v", e.Err) }
+func (e ErrRenewFailed) Unwrap() error { return e.Err }
+func (ErrRenewFailed) Op() Op          { return OpRenew }
+
+// ErrRecreateFailed wraps a failed CreatePullPointSubscription. Consumers
+// seeing this repeatedly should consider the camera offline.
+type ErrRecreateFailed struct{ Err error }
+
+func (e ErrRecreateFailed) Error() string { return fmt.Sprintf("recreate pull point: %v", e.Err) }
+func (e ErrRecreateFailed) Unwrap() error { return e.Err }
+func (ErrRecreateFailed) Op() Op          { return OpRecreate }

event/stream/types_test.go

@@ -0,0 +1,143 @@
+package stream
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestKindString(t *testing.T) {
+	tests := []struct {
+		name string
+		kind Kind
+		want string
+	}{
+		{"unknown", KindUnknown, "Unknown"},
+		{"motion", KindMotion, "Motion"},
+		{"tampering", KindTampering, "Tampering"},
+		{"image_quality", KindImageQuality, "ImageQuality"},
+		{"digital_input", KindDigitalInput, "DigitalInput"},
+		{"digital_output", KindDigitalOutput, "DigitalOutput"},
+		{"object_detected", KindObjectDetected, "ObjectDetected"},
+		{"audio_alarm", KindAudioAlarm, "AudioAlarm"},
+		{"out_of_range", Kind(255), "Kind(255)"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, tc.kind.String())
+		})
+	}
+}
+
+func TestKindStringsAreUnique(t *testing.T) {
+	seen := map[string]Kind{}
+	for k := KindUnknown; k <= KindAudioAlarm; k++ {
+		s := k.String()
+		prev, dup := seen[s]
+		assert.False(t, dup, "duplicate String %q for Kind(%d) and Kind(%d)", s, prev, k)
+		seen[s] = k
+	}
+}
+
+func TestStateString(t *testing.T) {
+	tests := []struct {
+		name  string
+		state State
+		want  string
+	}{
+		{"unknown", StateUnknown, "Unknown"},
+		{"active", StateActive, "Active"},
+		{"inactive", StateInactive, "Inactive"},
+		{"out_of_range", State(255), "State(255)"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, tc.state.String())
+		})
+	}
+}
+
+func TestPropertyOperationString(t *testing.T) {
+	tests := []struct {
+		name string
+		op   PropertyOperation
+		want string
+	}{
+		{"unknown", PropertyUnknown, "Unknown"},
+		{"initialized", PropertyInitialized, "Initialized"},
+		{"changed", PropertyChanged, "Changed"},
+		{"deleted", PropertyDeleted, "Deleted"},
+		{"out_of_range", PropertyOperation(255), "PropertyOperation(255)"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, tc.op.String())
+		})
+	}
+}
+
+func TestEventZeroValue(t *testing.T) {
+	var e Event
+	assert.Equal(t, KindUnknown, e.Kind)
+	assert.Equal(t, StateUnknown, e.State)
+	assert.Equal(t, PropertyUnknown, e.Operation)
+	assert.Empty(t, e.DeviceID)
+	assert.Nil(t, e.Source)
+	assert.Nil(t, e.Data)
+	assert.Empty(t, e.Topic)
+	assert.True(t, e.Timestamp.IsZero())
+	assert.True(t, e.DeviceTime.IsZero())
+}
+
+func TestEventFieldAssignmentRoundTrip(t *testing.T) {
+	now := time.Now().UTC()
+	deviceTime := now.Add(-2 * time.Second)
+	e := Event{
+		Kind:       KindMotion,
+		State:      StateActive,
+		Operation:  PropertyChanged,
+		DeviceID:   "axis-camera-01",
+		Source:     map[string]string{"InputToken": "DI1"},
+		Data:       map[string]string{"LogicalState": "true"},
+		Topic:      "tns1:Device/Trigger/DigitalInput",
+		Timestamp:  now,
+		DeviceTime: deviceTime,
+	}
+	assert.Equal(t, KindMotion, e.Kind)
+	assert.Equal(t, StateActive, e.State)
+	assert.Equal(t, PropertyChanged, e.Operation)
+	assert.Equal(t, "axis-camera-01", e.DeviceID)
+	assert.Equal(t, "DI1", e.Source["InputToken"])
+	assert.Equal(t, "true", e.Data["LogicalState"])
+	assert.Equal(t, "tns1:Device/Trigger/DigitalInput", e.Topic)
+	assert.True(t, e.Timestamp.Equal(now))
+	assert.True(t, e.DeviceTime.Equal(deviceTime))
+}
+
+// --- Typed errors -----------------------------------------------------
+
+func TestTypedErrors_UnwrapAndOp(t *testing.T) {
+	inner := errors.New("boom")
+	tests := []struct {
+		name string
+		err  error
+		op   Op
+	}{
+		{"pull", ErrPullFailed{Err: inner}, OpPull},
+		{"renew", ErrRenewFailed{Err: inner}, OpRenew},
+		{"recreate", ErrRecreateFailed{Err: inner}, OpRecreate},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.True(t, errors.Is(tc.err, inner), "errors.Is should unwrap to inner")
+			assert.Contains(t, tc.err.Error(), "boom")
+			if e, ok := tc.err.(interface{ Op() Op }); ok {
+				assert.Equal(t, tc.op, e.Op())
+			} else {
+				t.Fatalf("%T does not expose Op()", tc.err)
+			}
+		})
+	}
+}

event/types.go

@@ -133,6 +133,7 @@ type MessageBody struct {
 
 type MessageDescription struct {
 	PropertyOperation xsd.AnyType `xml:"PropertyOperation,attr"`
+	UtcTime           xsd.AnyType `xml:"UtcTime,attr"`
 	Source            Source      `json:",omitempty" xml:",omitempty"`
 	Data              Data        `json:",omitempty" xml:",omitempty"`
 }

examples/event/stream/main.go

@@ -0,0 +1,176 @@
+// Command streamtest opens an event stream against an ONVIF camera and
+// prints decoded events as they arrive. Useful for verifying the
+// classifier against real-camera topics; not intended as a production
+// tool.
+//
+// # Usage
+//
+//	go run ./examples/event/stream \
+//	  -xaddr 192.168.1.10 \
+//	  -username root \
+//	  -duration 60s
+//
+// # Credentials
+//
+// The camera password is read, in order of preference:
+//
+//  1. The ONVIF_PASSWORD environment variable.
+//  2. A file pointed at by -password-file (newline stripped).
+//  3. Interactive prompt when stdin is a tty.
+//
+// -password is also accepted but DISCOURAGED — it leaks the credential
+// into shell history and the system process listing. Use only for
+// throwaway dev cameras.
+package main
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/kerberos-io/onvif"
+	"github.com/kerberos-io/onvif/event/stream"
+)
+
+func main() {
+	xaddr := flag.String("xaddr", "", "camera host or host:port (required)")
+	username := flag.String("username", "", "ONVIF user (required)")
+	insecurePassword := flag.String("password", "", "INSECURE — leaks into shell history; prefer ONVIF_PASSWORD env or -password-file")
+	passwordFile := flag.String("password-file", "", "read password from this file (newline trimmed)")
+	deviceID := flag.String("device-id", "", "logical name printed with each event (default: xaddr)")
+	filter := flag.String("filter", "", "raw ONVIF ConcreteSet topic filter (empty = all topics, works on AXIS)")
+	pullTimeout := flag.Duration("pull-timeout", 5*time.Second, "server-side wait per PullMessages call")
+	duration := flag.Duration("duration", 0, "stop after this long (0 = run until Ctrl-C)")
+	flag.Parse()
+
+	if *xaddr == "" || *username == "" {
+		flag.Usage()
+		os.Exit(2)
+	}
+	if *deviceID == "" {
+		*deviceID = *xaddr
+	}
+
+	password, err := loadPassword(*insecurePassword, *passwordFile)
+	if err != nil {
+		log.Fatalf("password: %v", err)
+	}
+
+	dev, err := onvif.NewDevice(onvif.DeviceParams{
+		Xaddr:    *xaddr,
+		Username: *username,
+		Password: password,
+		AuthMode: onvif.UsernameTokenAuth,
+	})
+	if err != nil {
+		log.Fatalf("connect: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if *duration > 0 {
+		var done context.CancelFunc
+		ctx, done = context.WithTimeout(ctx, *duration)
+		defer done()
+	}
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-sigs
+		cancel()
+	}()
+
+	s, err := stream.NewStream(ctx, dev, stream.Options{
+		DeviceID:       *deviceID,
+		RawTopicFilter: *filter,
+		PullTimeout:    *pullTimeout,
+	})
+	if err != nil {
+		log.Fatalf("open stream: %v", err)
+	}
+	defer func() {
+		if err := s.Close(); err != nil {
+			log.Printf("stream close: %v", err)
+		}
+	}()
+
+	log.Printf("streaming from %s (device-id=%s, filter=%q)", *xaddr, *deviceID, *filter)
+	for {
+		select {
+		case <-ctx.Done():
+			log.Printf("done (%v)", ctx.Err())
+			return
+		case ev, ok := <-s.Events():
+			if !ok {
+				return
+			}
+			fmt.Printf("%s  kind=%-15s state=%-9s op=%-12s topic=%s",
+				ev.Timestamp.Format(time.RFC3339), ev.Kind, ev.State, ev.Operation, ev.Topic)
+			if ev.AfterReconnect {
+				fmt.Print("  [after-reconnect]")
+			}
+			if len(ev.Source) > 0 {
+				fmt.Printf("  source=%v", ev.Source)
+			}
+			if len(ev.Data) > 0 {
+				fmt.Printf("  data=%v", ev.Data)
+			}
+			fmt.Println()
+		case e, ok := <-s.Errors():
+			if !ok {
+				return
+			}
+			var pull stream.ErrPullFailed
+			var recreate stream.ErrRecreateFailed
+			switch {
+			case errors.As(e, &recreate):
+				log.Printf("RECREATE failed: %v (camera may be offline)", recreate.Err)
+			case errors.As(e, &pull):
+				log.Printf("pull error (will retry): %v", pull.Err)
+			default:
+				log.Printf("stream error: %v", e)
+			}
+		}
+	}
+}
+
+// loadPassword resolves the camera password from the environment first
+// (ONVIF_PASSWORD), then -password-file, then an interactive prompt as
+// a last resort. The insecure -password flag is honoured only if
+// nothing else is set, and a warning is logged.
+func loadPassword(insecure, file string) (string, error) {
+	if env := os.Getenv("ONVIF_PASSWORD"); env != "" {
+		return env, nil
+	}
+	if file != "" {
+		b, err := os.ReadFile(file)
+		if err != nil {
+			return "", fmt.Errorf("read %s: %w", file, err)
+		}
+		return strings.TrimRight(string(b), "\r\n"), nil
+	}
+	if insecure != "" {
+		log.Print("WARNING: -password leaks into shell history and process listings; prefer ONVIF_PASSWORD env or -password-file")
+		return insecure, nil
+	}
+	// Interactive prompt — works when stdin is a tty. We use a plain
+	// reader (rather than golang.org/x/term hidden input) to keep
+	// this example dependency-free; in production, callers should
+	// integrate term.ReadPassword.
+	fmt.Fprint(os.Stderr, "ONVIF password (visible): ")
+	r := bufio.NewReader(os.Stdin)
+	line, err := r.ReadString('\n')
+	if err != nil {
+		return "", errors.New("no password supplied (set ONVIF_PASSWORD, -password-file, or pipe input)")
+	}
+	return strings.TrimRight(line, "\r\n"), nil
+}